﻿#include "hnode.h"
#include "uv.h"
#include "queue.h"
#include "uv-common.h"
#include "uv/tree.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <stdio.h>
#include <errno.h> 
#include <conio.h>
#include <stdlib.h>
#include <direct.h>
#include <time.h>
#include <share.h>
#include <stdarg.h>
#include <string.h>
#ifdef _WIN32
#include <windows.h>
#endif
#pragma comment(lib, "IPHLPAPI.lib")
#pragma comment(lib, "Psapi.lib")
#pragma comment(lib, "userenv.lib")
#define MAX_LOG_LEN (1024)
#define MAX_LOG_PAHT (256)
#define MAX_LOG_POOL_ITMES (128)
#define MAX_LOG_LINES (102400)
struct hbuffer
{
    hsize read_pos;
    hsize write_pos;
    hsize capacity;
    char* buffer;
};
#define MAX_PATH_LEN (260)

static int create_dir(const char* paht) {
    if (_mkdir(paht) == 0) {
        return 1;
    } else {
        return 0;
    }
}

static int remove_dir(const char* path) {
    if (_rmdir(path) == 0) {
        return 1;
    }
    return 0;
}

int get_current_dir(char* buf, int len) {
    if (_fullpath(buf, "", len) != NULL) {
        return 1;
    }
    return 0;
}

int is_dir_exist(const char* path) {
    char full_path[MAX_PATH_LEN] = {0};
    if (_fullpath(full_path, path, MAX_PATH_LEN) == NULL) {
        return 0;
    }
    struct _stat buf;
    int result = _stat(full_path, &buf);
    if (result != 0) {
        return 0;
    } else {
        if (buf.st_mode & _S_IFDIR) {
            return 1;
        }
    }
    return 0;
}

int create_dir_recursive(const char* path) {
    char buf_path[MAX_PATH_LEN] = {0};
    if (_fullpath(buf_path, path, MAX_PATH_LEN) == NULL) {
        return 0;
    }
    size_t len = strlen(buf_path);
    char buf_path_copy[MAX_PATH_LEN] = {0};
    for (size_t i = 0; i < len; i++) {
        buf_path_copy[i] = buf_path[i];
        if (buf_path[i] == '\\' || buf_path[i] == '/') {
            if (!is_dir_exist(buf_path_copy)) {
                int ret = create_dir(buf_path_copy);
                if (ret == 0) {
                    printf("create_dir_recursive for[%s]fail while create sub directory[%s]\n", buf_path, buf_path_copy);
                    return 0;
                }
            }
        }
    }
    if (!is_dir_exist(buf_path)) {
        create_dir(buf_path);
    }
    return 1;
}

static int do_remove_dir_recursive(const char* folder_path) {
    char buf_find_file[MAX_PATH_LEN] = {0};
    sprintf_s(buf_find_file, sizeof(buf_find_file), "%s\\*", folder_path);
    HANDLE hFind;
    WIN32_FIND_DATAA findData;
    hFind = FindFirstFileA(buf_find_file, &findData);
    if (hFind == INVALID_HANDLE_VALUE) {
        return 0;
    }
    do {
        if (strcmp(findData.cFileName, ".") == 0 || strcmp(findData.cFileName, "..") == 0) {
            continue;
        }
        if (findData.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) {
            char buf_new_folder[MAX_PATH_LEN] = {0};
            sprintf_s(buf_new_folder, sizeof(buf_new_folder), "%s\\%s", folder_path, findData.cFileName);
            do_remove_dir_recursive(buf_new_folder);
        } else {
            char to_delete_file[MAX_PATH_LEN] = {0};
            sprintf_s(to_delete_file, sizeof(to_delete_file), "%s\\%s", folder_path, findData.cFileName);
            int ret = remove(to_delete_file);
            if (ret != 0) {
                DWORD err = GetLastError();
                printf("remove file [%s]failed [%d]\n", to_delete_file, err);
            }
        }
    } while (FindNextFileA(hFind, &findData));
    remove_dir(folder_path);
    FindClose(hFind);
    return 1;
}
//use it carefully !!! please
int remove_dir_recursive(const char* path) {
    if (!is_dir_exist(path)) {
        return 0;
    }
    do_remove_dir_recursive(path);
    return 1;
}

void sync_c_log(const char* file_name, const char* fmt, ...) {
    static int swap_times = 0;
    time_t rawtime = 0;
    struct tm tm_info;
    char time_buf[32] = {0};
    time(&rawtime);
    localtime_s(&tm_info, &rawtime);
    strftime(time_buf, sizeof time_buf, "%Y-%m-%e_%H-%M-%S", &tm_info);
    char log_path[MAX_LOG_PAHT + 32] = {0};
    get_current_dir(log_path, MAX_LOG_PAHT);
    sprintf_s(log_path + strlen(log_path), (sizeof log_path) - strlen(log_path), "\\sync_c_log");
    if (!is_dir_exist(log_path)) {
        create_dir_recursive(log_path);
    }
    char file_path[MAX_LOG_PAHT + 32] = {0};
    //sprintf_s(file_path, sizeof file_path, "%s\\%s_%s_%d.log", log_path, file_name, time_buf, swap_times++);
    DWORD processId = GetCurrentProcessId();//当前进程id
    sprintf_s(file_path, sizeof file_path, "%s\\%s_processid_%d.log", log_path, file_name, processId);
    FILE* file = _fsopen(file_path, "a+", _SH_DENYNO);
    if (file == NULL) {
        auto b = GetLastError();
        auto debug = 0;
        debug++;
        printf("open file failed %s \n", file_path);
        return;
    }
    char log_content[MAX_LOG_PAHT + 32] = {0};
    va_list args;
    va_start(args, fmt);
    vsnprintf(log_content, sizeof(log_content), fmt, args);
    va_end(args);
    fseek(file, 0L, SEEK_END);

    time_t tsec = time(NULL);
    unsigned long long ticm = GetTickCount64();
    int ms = (int)(ticm % 1000);
    char bufTime[128] = {0};
    sprintf_s(bufTime, sizeof bufTime, "%llu.%d  ", tsec, ms);
    size_t sizew0 = fwrite(bufTime, 1, strlen(bufTime), file);
    size_t sizew = fwrite(log_content, 1, strlen(log_content), file);
    fflush(file);
    fclose(file);
}

//#define SYNC_CLOSE_lOG(fmt, ...)  sync_c_log("sync_log" ,fmt, ##__VA_ARGS__)
//#define SYNC_SET_STATE_lOG(fmt, ...)  sync_c_log("set_states" ,fmt, ##__VA_ARGS__)
//#define SYNC_SHUTDOWN_lOG(fmt, ...) sync_c_log("shutdown" ,fmt, ##__VA_ARGS__)
//#define SYNC_PACKAGE_lOG(fmt, ...) sync_c_log("package_cache" ,fmt, ##__VA_ARGS__)

#define SYNC_CLOSE_lOG(fmt, ...)  
#define SYNC_SET_STATE_lOG(fmt, ...) 
#define SYNC_SHUTDOWN_lOG(fmt, ...) 
#define SYNC_PACKAGE_lOG(fmt, ...) 


#define ASSERT(expr)                                      \
 do {                                                     \
  if (!(expr)) {                                          \
    HLOG_ERROR(                                           \
            "Assertion failed in %s on line %d: %s\n",    \
            __FILE__,                                     \
            __LINE__,                                     \
            #expr);                                       \
  }                                                       \
 } while (0)

struct log_item
{
    QUEUE queue_item;
    e_log_level log_level;
    time_t log_time;
    int ms;
    int console;
    char log_content[MAX_LOG_LEN];
};

struct one_hlog_file
{
    uv_mutex_t mutex;
    QUEUE log_items;
    uv_thread_t thread_id;
    uv_mutex_t pool_mutex;
    QUEUE log_item_pool;
    int stop;
    FILE* file;
    int lines;
    char path[MAX_LOG_PAHT];
    int swap_times;
    e_log_level min_level;
    RB_ENTRY(one_hlog_file) node;
    QUEUE sq;
    char buf_hold[MAX_LOG_LEN + 128];
};

static int log_file_cmp(struct one_hlog_file* left, struct one_hlog_file* right) {
    int result = strcmp(left->path, right->path);
    return result;
}

RB_HEAD(FILE_MAP, one_hlog_file);
RB_GENERATE_STATIC(FILE_MAP, one_hlog_file, node, log_file_cmp);

struct h_log_system
{
    uv_rwlock_t rwlock;
    struct FILE_MAP file_map;
    e_log_level min_level;
    char root[MAX_LOG_PAHT];
    uv_key_t tls_find;
};

static struct h_log_system* ph_log = NULL;

static const char* get_log_level_name(e_log_level l) {
    switch (l) {
    case LOG_DEBUG:
    {
        return "DEBUG";
    }break;
    case LOG_INFO:
    {
        return "INFO";
    }break;
    case LOG_WARNING:
    {
        return "WARNING";
    }break;
    case LOG_ERROR:
    {
        return "ERROR";

    }break;
    default:
    break;
    }
    return "UNKNOWN";
}

static int queue_len(struct one_hlog_file* file, QUEUE* q) {
    uv_mutex_lock(&file->mutex);
    int ret = 0;
    QUEUE* t;
    QUEUE_FOREACH(t, q) {
        struct log_item* item = QUEUE_DATA(t, struct log_item, queue_item);
        printf("item [%d][%p][%s]\n", ret, item, item->log_content);
        ret++;
    }
    uv_mutex_unlock(&file->mutex);
    return ret;
}

static void add_log_item(struct one_hlog_file* plog, struct log_item* it) {
    uv_mutex_lock(&plog->mutex);
    QUEUE_INSERT_TAIL(&plog->log_items, &it->queue_item);
    uv_mutex_unlock(&plog->mutex);
}

static void try_relog(struct one_hlog_file* plog) {
    int new_file = 0;
    if (plog->file == NULL) {
        new_file = 1;
    } else {
        if (plog->lines >= MAX_LOG_LINES) {
            new_file = 1;
            plog->lines = 0;
        }
    }
    if (new_file) {
        if (plog->file) {
            fflush(plog->file);
            fclose(plog->file);
            plog->file = NULL;
        }
        time_t rawtime = 0;
        struct tm tm_info;
        char time_buf[32] = {0};
        time(&rawtime);
        localtime_s(&tm_info, &rawtime);
        strftime(time_buf, sizeof time_buf, "%Y-%m-%e_%H-%M-%S", &tm_info);
        char log_path[MAX_LOG_PAHT + 32] = {0};
        get_current_dir(log_path, MAX_LOG_PAHT);
        sprintf_s(log_path + strlen(log_path), (sizeof log_path) - strlen(log_path), "\\%s", ph_log->root);
        if (!is_dir_exist(log_path)) {
            create_dir_recursive(log_path);
        }
        char file_path[MAX_LOG_PAHT + 32] = {0};
        sprintf_s(file_path, sizeof file_path, "%s\\%s_%s_%d.log", log_path, plog->path, time_buf, plog->swap_times++);
        plog->file = _fsopen(file_path, "a+", _SH_DENYNO);
        if (plog->file == NULL) {
            auto b = GetLastError();
            auto debug = 0;
            debug++;
        }
    }
}

static void log_work(void* arg) {
    struct one_hlog_file* plog = (struct one_hlog_file*)(arg);
    while (!(plog->stop)) {

        QUEUE_INIT(&plog->sq);
        uv_mutex_lock(&plog->mutex);
        QUEUE_MOVE(&plog->log_items, &plog->sq);
        uv_mutex_unlock(&plog->mutex);
        if (!QUEUE_EMPTY(&plog->sq)) {
            QUEUE* pq = NULL;
            while (!QUEUE_EMPTY(&plog->sq)) {
                pq = QUEUE_NEXT(&plog->sq);
                QUEUE_REMOVE(pq);
                struct log_item* plog_item = QUEUE_DATA(pq, struct log_item, queue_item);
                if (plog_item->log_level < plog->min_level) {
                    continue;
                }
                try_relog(plog);
                struct tm tm_info;
                char time_buf[32] = {0};
                //time(&plog_item->log_time);
                localtime_s(&tm_info, &plog_item->log_time);
                strftime(time_buf, sizeof time_buf, "%Y-%m-%e_%H:%M:%S", &tm_info);
                // char buf_hold[MAX_LOG_LEN + 128] = {0};
                memset(plog->buf_hold, 0, sizeof plog->buf_hold);
                const char* plevel_name = get_log_level_name(plog_item->log_level);
                sprintf_s(plog->buf_hold, sizeof(plog->buf_hold), "%s.%d %s %s", time_buf, plog_item->ms, plevel_name, plog_item->log_content);
                if (plog_item->console) {
                    if (plog_item->log_level == LOG_WARNING) {
                        CONSOLE_SCREEN_BUFFER_INFO csbiInfo;
                        HANDLE h = GetStdHandle(STD_OUTPUT_HANDLE);
                        GetConsoleScreenBufferInfo(h, &csbiInfo);
                        DWORD d = csbiInfo.wAttributes;
                        SetConsoleTextAttribute(h,
                                                FOREGROUND_RED |
                                                FOREGROUND_GREEN |
                                                FOREGROUND_INTENSITY);

                        printf("%s", plog->buf_hold);
                        SetConsoleTextAttribute(h, d);

                    } else if (plog_item->log_level == LOG_ERROR) {
                        CONSOLE_SCREEN_BUFFER_INFO csbiInfo;
                        HANDLE h = GetStdHandle(STD_OUTPUT_HANDLE);
                        GetConsoleScreenBufferInfo(h, &csbiInfo);
                        DWORD d = csbiInfo.wAttributes;
                        SetConsoleTextAttribute(h,
                                                FOREGROUND_RED |
                                                FOREGROUND_INTENSITY);

                        printf("%s", plog->buf_hold);
                        SetConsoleTextAttribute(h, d);
                    } else {
                        printf("%s", plog->buf_hold);//todo add color here 

                    }
                }
                fseek(plog->file, 0L, SEEK_END);
                size_t sizew = fwrite(plog->buf_hold, 1, strlen(plog->buf_hold), plog->file);
                plog->lines++;
                free(plog_item);
            }
            fflush(plog->file);
            /*uv_mutex_lock(&plog->pool_mutex);
            QUEUE_ADD(&plog->log_item_pool, &sq);
            uv_mutex_unlock(&plog->pool_mutex);*/
            //free(sq);
        } else {
            Sleep(1);
        }
    }
}

struct one_hlog_file* create_hlog_file(e_log_level min_allowed_log_level, const char* path) {
    struct one_hlog_file* pfile = NULL;
    pfile = malloc(sizeof(*pfile));
    memset(pfile, 0, sizeof(*pfile));
    QUEUE_INIT(&pfile->log_items);
    uv_mutex_init(&pfile->mutex);
    QUEUE_INIT(&pfile->log_item_pool);
    uv_mutex_init(&pfile->pool_mutex);
    for (int i = 0; i < MAX_LOG_POOL_ITMES; i++) {
        struct log_item* p = malloc(sizeof(struct log_item));
        memset(p, 0, sizeof(*p));
        QUEUE_INIT(&p->queue_item);
        QUEUE_INSERT_HEAD(&pfile->log_item_pool, &p->queue_item);
    }
    pfile->file = NULL;
    pfile->lines = 0;
    pfile->stop = 0;
    pfile->swap_times = 0;
    pfile->min_level = min_allowed_log_level;
    /* size_t len = strlen(ph_log->root);
     if (len <= 0) {
         sprintf_s(pfile->path, sizeof pfile->path, "%s", path);
     } else {
         if (ph_log->root[len - 1] == '\\') {
             sprintf_s(pfile->path, sizeof pfile->path, "%s%s", ph_log->root, path);
         } else {
             sprintf_s(pfile->path, sizeof pfile->path, "%s\\%s", ph_log->root, path);
         }
     }*/
    strcpy_s(pfile->path, MAX_LOG_PAHT, path);
    int r = uv_thread_create(&pfile->thread_id, log_work, pfile);
    ASSERT(r == 0);
    return pfile;
}

void stop_hlog_file(struct one_hlog_file* pfile) {
    pfile->stop = 1;
    uv_thread_join(&pfile->thread_id);
    uv_mutex_destroy(&pfile->mutex);
    uv_mutex_destroy(&pfile->pool_mutex);
    /*while (QUEUE_EMPTY(&pfile->log_item_pool)) {
        QUEUE* h = QUEUE_HEAD(&pfile->log_item_pool);
        struct log_item* i = QUEUE_DATA(h, struct log_item, queue_item);
        QUEUE_REMOVE(h);
        free(i);
    }*/
    free(pfile);
}

void init_hlog(e_log_level min_allowed_log_level, const char* path) {
    if (ph_log) {
        return;
    }
    ph_log = malloc(sizeof(*ph_log));
    memset(ph_log, 0, sizeof(*ph_log));
    RB_INIT(&ph_log->file_map);
    int r = uv_rwlock_init(&ph_log->rwlock);
    strcpy_s(ph_log->root, MAX_LOG_PAHT, path);
    ph_log->min_level = min_allowed_log_level;
    ASSERT(0 == uv_key_create(&ph_log->tls_find));
}

void uninit_hlog() {
    if (!ph_log) {
        return;
    }
    uv_rwlock_wrlock(&ph_log->rwlock);
    while (!RB_EMPTY(&ph_log->file_map)) {
        struct one_hlog_file* tmp = RB_ROOT(&ph_log->file_map);
        FILE_MAP_RB_REMOVE(&ph_log->file_map, RB_ROOT(&ph_log->file_map));
        stop_hlog_file(tmp);
    }
    uv_rwlock_wrunlock(&ph_log->rwlock);
    uv_key_delete(&ph_log->tls_find);
}

static struct log_item* get_log_item(struct one_hlog_file* file) {
    struct log_item* p = NULL;
    // uv_mutex_lock(&file->pool_mutex);
     // if (QUEUE_EMPTY(&file->log_item_pool)) {
    p = malloc(sizeof(struct log_item));
    memset(p, 0, sizeof(struct log_item));
    QUEUE_INIT(&p->queue_item);
    /*} else {
        QUEUE* h = QUEUE_HEAD(&file->log_item_pool);
        p = QUEUE_DATA(h, struct log_item, queue_item);
        QUEUE_REMOVE(h);
        memset(p, 0, sizeof(struct log_item));
        QUEUE_INIT(&p->queue_item);
    }*/
    // uv_mutex_unlock(&file->pool_mutex);
    return p;
}

void hlog(const char* tag, e_log_level l, int console1, const char* fmt, ...) {
    if (!ph_log) {
        ASSERT(0);
    }
    if (l < ph_log->min_level) {
        return;
    }

    struct one_hlog_file* p = (struct one_hlog_file*)uv_key_get(&ph_log->tls_find);
    if (p == NULL) {
        p = malloc(sizeof(*p));
        uv_key_set(&ph_log->tls_find, p);
    }

    strcpy_s(p->path, sizeof p->path, tag);
    uv_rwlock_rdlock(&ph_log->rwlock);
    struct one_hlog_file* pfind = FILE_MAP_RB_FIND(&ph_log->file_map, p);
    uv_rwlock_rdunlock(&ph_log->rwlock);

    if (pfind == NULL) {
        pfind = create_hlog_file(ph_log->min_level, tag);
        uv_rwlock_wrlock(&ph_log->rwlock);
        FILE_MAP_RB_INSERT(&ph_log->file_map, pfind);
        uv_rwlock_wrunlock(&ph_log->rwlock);
    }

    struct log_item* plog_item = get_log_item(pfind);
    plog_item->log_time = time(NULL);
    unsigned long long ticm = GetTickCount64();
    plog_item->ms = (ticm % 1000);
    plog_item->console = console1;
    plog_item->log_level = l;
    va_list args;
    va_start(args, fmt);
    //vsprintf_s(plog_item->log_content, sizeof(plog_item->log_content), fmt, args);
    //vsprintf_s是不安全函数 恶心 fuck it 
    vsnprintf(plog_item->log_content, sizeof(plog_item->log_content), fmt, args);
    va_end(args);
    add_log_item(pfind, plog_item);
}



//后续考虑变成整就是一块内存的 存C语言的方式 
struct hbuffer* malloc_hbuffer(hsize size) {
    struct hbuffer* phb = malloc(sizeof(struct hbuffer));
    memset(phb, 0, sizeof(struct hbuffer));
    phb->buffer = malloc(size);
    phb->capacity = size;
    return phb;
}

void free_hbuffer(struct hbuffer* phb) {
    free(phb->buffer);
    free(phb);
}

hsize write_data(struct hbuffer* phb, const char* data, hsize len) {
    hsize write_cap = phb->capacity - phb->write_pos;
    hsize total_free_cap = write_cap + phb->read_pos;
    hsize data_avail = phb->write_pos - phb->read_pos;

    if (write_cap >= len) {
        memcpy_s((phb->buffer + phb->write_pos), write_cap, data, len);
        phb->write_pos += len;
    } else {
        if (total_free_cap >= len) {
            memmove_s(phb->buffer, data_avail, (phb->buffer + phb->read_pos), data_avail);
            memcpy_s((phb->buffer + data_avail), len, data, len);
            phb->write_pos = data_avail + len;
            phb->read_pos = 0;
        } else {
            hsize newLen = len + phb->capacity;
            char* p = malloc(newLen);
            memset(p, 0, newLen);
            memcpy_s(p, newLen, phb->buffer + phb->read_pos, data_avail);
            memcpy_s(p + data_avail, newLen - data_avail, data, len);
            phb->read_pos = 0;
            phb->write_pos = data_avail + len;
            phb->capacity = newLen;
            free(phb->buffer);
            phb->buffer = p;
        }
    }

    return 0;
}

void advance_write_pos(struct hbuffer* phb, hsize pos) {
    phb->write_pos += pos;
}

void advance_read_pos(struct hbuffer* phb, hsize pos) {
    phb->read_pos += pos;
}

char* get_readable_pos(struct hbuffer* phb) {
    return phb->buffer + phb->read_pos;
}

char* get_writeable_pos(struct hbuffer* phb) {
    return phb->buffer + phb->write_pos;
}

hsize get_readable_num(struct hbuffer* phb) {
    return  phb->write_pos - phb->read_pos;
}

hsize get_writeable_num(struct hbuffer* phb) {
    return  phb->capacity - phb->write_pos;
}


#define GUARD_TIMEER_GAP_MS (5)
//BIT_NUM must be at least 1 & max 32
#define BIT_NUM (4)
#define THREAD_NUM (1 << BIT_NUM)




#define NET_APP_CMD_MAP(XX)                                                 \
    XX(ON_CONNECT_SUCCESS,ON_CONNECT_SUCCESS)                               \
    XX(ON_CONNECT_FAIL,ON_CONNECT_FAIL)                                     \
    XX(ON_CONNECTION,ON_CONNECTION)                                         \
    XX(ON_RECEIVE,ON_RECEIVE)                                               \
    XX(ON_CLOSE,ON_CLOSE)                                                   \

typedef enum
{
#define XX(uc,lc) E_NET_APP_##uc,
    NET_APP_CMD_MAP(XX)
#undef XX 
    E_NET_APP_ON_MAX
}e_net_app_cmd;

#define NET_APP_CMD_PRIVATE_FIELDS                                          \
void* event_node[2];                                                        \
e_net_app_cmd net_app_type;                                                 \
int index;                                                                  \
connection_id net_object_id;                                                \
void* return_source[2];                                                     \
phnet_ev_rv ev_rv;

#define IP_PORT_FIELDS                                                      \
char remote_ip[MAX_IP_LEN];                                                 \
int remote_port;                                                            \
char local_ip[MAX_IP_LEN];                                                  \
int local_port;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
}net_app_base;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
        IP_PORT_FIELDS
        char name[MAX_NET_OBJECT_ID_LEN];
}net_app_on_connect_success;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
        char name[MAX_NET_OBJECT_ID_LEN];

}net_app_on_connect_fail;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
        IP_PORT_FIELDS
        char name[MAX_NET_OBJECT_ID_LEN];
}net_app_on_connection;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
        char* buf;
    int len;
}net_app_on_receive;

typedef struct
{
    NET_APP_CMD_PRIVATE_FIELDS
}net_app_on_close;

typedef void (*app_handle_func)(uv_loop_t* ploop, struct uv_app_s* ptr);

#define CMP_TS(FN ,TYPE ,MEMBER) \
static int FN(struct TYPE* left, struct TYPE* right) {\
int result = strcmp(left->MEMBER, right->MEMBER);\
return result;\
}

#define CMP_TI(FN ,TYPE ,MEMBER)\
static int FN(struct TYPE* left, struct TYPE* right) {\
    if (left->MEMBER == right->MEMBER) {\
        return 0;\
    } else {\
        if (left->MEMBER > right->MEMBER) {\
            return 1;\
        }\
    }\
    return -1;\
}

#define OP_TYPE_MAP(XX)                                   \
  XX(CONNECT,CONNECT )                                    \
  XX(ACCEPT,ACCEPT )                                      \
  XX(SEND, SEND)                                          \
  XX(CLOSE, CLOSE)                                        \
  XX(STOP_CONNECTOR, STOP_CONNECTOR)                      \
  XX(STOP_ACCEPTOR, STOP_ACCEPTOR)                        \
  XX(STOP_ALL, STOP_ALL)                                  \
  XX(HB, HB)                                              \

typedef enum
{
#define XX(uc, lc) OP_##uc,
    OP_TYPE_MAP(XX)
#undef XX 
    OP_MAX
} e_op_type;

const char* get_op_name(e_op_type e) {
    switch (e) {
#define XX(uc,lc) case OP_##uc: return "OP_"#lc;
        OP_TYPE_MAP(XX)
#undef XX 
    default:
    break;
    }
    return "OP_???";
}

#define POST_TYPE_MAP(XX)                                                   \
XX(NONE,NONE)                                                               \
XX(CONNECTING, CONNECTING)                                                  \
XX(CONNECT_FAIL_WAIT_TIMEOUT, CONNECT_FAIL_WAIT_TIMEOUT)                    \
XX(TIMER_RECONNECTING, TIMER_RECONNECTING)                                  \
XX(WRITEING, WRITEING)                                                      \
XX(READING, READING)                                                        \
XX(SHUTDOWNING, SHUTDOWNING)                                                \
XX(CLOSEING, CLOSEING)                                                      \

typedef enum
{
#define XX(uc,lc) POST_##uc,
    POST_TYPE_MAP(XX)
#undef  XX 
    POST_MAX
} e_post_type;

int get_post_mask(e_post_type e) {
    const static int static_post_mask[POST_MAX] = {
#define XX(uc,lc) 1 << POST_##uc,
    POST_TYPE_MAP(XX)
#undef  XX
    };
    return static_post_mask[e];
}

const char* get_posted_name(e_post_type e) {
    switch (e) {
#define XX(uc,lc) case POST_##uc: return "POST_"#lc;
        POST_TYPE_MAP(XX)
#undef XX 
    default:
    break;
    }
    return "POST_???";
}

#define POST_CLOSING_REASON_MAP(XX)                                          \
XX(READ_ERROR,READ_ERROR)                                                    \
XX(WRITE_ERROR,WRITE_ERROR)                                                  \
XX(CONNECT_ERROR,CONNECT_ERROR)                                              \
XX(CONNECT_FAIL_NO_RECONNECT,CONNECT_FAIL_NO_RECONNECT)                      \
XX(ACTIVE,ACTIVE)                                                            \

typedef enum
{
#define XX(uc,lc) CLOSING_##uc,
    POST_CLOSING_REASON_MAP(XX)
#undef  XX 
    CLOSING_MAX
} e_closing_reason;

const char* get_closing_reason_name(e_closing_reason e) {
    switch (e) {
#define XX(uc,lc) case CLOSING_##uc: return "CLOSING_"#lc;
        POST_CLOSING_REASON_MAP(XX)
#undef XX 
    default:
    break;
    }
    return "POST_???";
}

struct connection_posted
{
    int post_type;
    int write_times;
    e_closing_reason reason;
};

static connection_id sbit_move = 1;

static const char* get_bit_maps_string(e_post_type t) {
    switch (t) {
#define XX(uc,lc) case POST_##uc : return "POST_"#lc;
        POST_TYPE_MAP(XX)
#undef XX
    default:
    break;
    }
    return "POST_???";
}

static void show_bit_maps(connection_id id, connection_id flag) {
#define MAX_BIT_MAP_STR_LEN (512)
    char buf[MAX_BIT_MAP_STR_LEN * 2] = {0};
    for (int i = 0; i < POST_MAX/* sizeof(connection_bit_maps) / sizeof(connection_id)*/; i++) {
        if (flag & (sbit_move << i)) {
            sprintf_s(buf + strlen(buf), MAX_BIT_MAP_STR_LEN, "%s\n", get_bit_maps_string(i));
        }
    }
    HLOG_INFO("show_bit_maps connection id [%llu] flags[%s]", id, buf);
}

const char* get_net_app_name(e_net_app_cmd op) {
    switch (op) {
#define XX(uc,lc) case E_NET_APP_##uc : return #lc;
        NET_APP_CMD_MAP(XX)
#undef XX  
    default:
    break;
    }
    return "E_NET_APP_???";
}
void init_app_handle();
void uninit_app_handle();

#define CONNECTION_STATE_MAP(XX)                            \
XX(INIT,INIT)                                               \
XX(CONNECTED,CONNECTED)                                     \
XX(CONNECTING,CONNECTING)                                   \
XX(CLOSING,CLOSING)                                         \
XX(CLOSED,CLOSED)                                           \

typedef enum
{
#define XX(uc,lc) STATE_##uc,
    CONNECTION_STATE_MAP(XX)
#undef XX
    STATE_MAX
} e_connection_state;

const char* get_connection_state_name(e_connection_state e) {
    switch (e) {
#define XX(uc,lc) case STATE_##uc : return #lc ;
        CONNECTION_STATE_MAP(XX)
#undef XX
    default:
    break;
    }
    return "STATE_???";
}

typedef enum
{
    LOOP_STATE_NORMAL = 0,
    LOOP_STATE_RUNNING,
    LOOP_STATE_CLOSED,
} e_loop_state;

struct net_object_string_to_id
{
    RB_ENTRY(net_object_string_to_id) node;
    char name[MAX_NET_OBJECT_ID_LEN];
    connection_id id;
};
static void on_connect(uv_connect_t* req, int status);

CMP_TS(net_object_string_cmp, net_object_string_to_id, name)
RB_HEAD(NET_OBJECT_STR2ID, net_object_string_to_id);
RB_GENERATE_STATIC(NET_OBJECT_STR2ID, net_object_string_to_id, node, net_object_string_cmp)

struct connection_meta
{
    connection_id cid;
    RB_ENTRY(connection_meta) node;
};

CMP_TI(connection_meta_cmp, connection_meta, cid)
RB_HEAD(CONNECTION_META_RBT, connection_meta);
RB_GENERATE_STATIC(CONNECTION_META_RBT, connection_meta, node, connection_meta_cmp)

struct app_handle
{
    RB_ENTRY(app_handle) node;
    app_handle_func handle;
    int type;
};

CMP_TI(app_handle_cmp, app_handle, type)
RB_HEAD(RB_APP_HANDLE, app_handle);
RB_GENERATE_STATIC(RB_APP_HANDLE, app_handle, node, app_handle_cmp)


#define APP_HANDLE_MAP(XX)                                          \
  XX(CONNECT,handle_op_connect )                                    \
  XX(ACCEPT,handle_op_accept )                                      \
  XX(SEND, handle_op_send)                                          \
  XX(CLOSE, handle_op_close)                                        \
  XX(STOP_CONNECTOR, handle_op_stop_connector)                      \
  XX(STOP_ACCEPTOR, handle_op_stop_acceptor)                        \
  XX(STOP_ALL, handle_op_stop_all)                                  \
  XX(HB, handle_op_hb)                                              \

struct app_handle* create_app_handle(e_op_type e);

//todo if multi logic thread add lock here 
typedef struct
{
    uv_write_t req;
    uv_buf_t buf;
} write_req_t;

typedef struct hbuf_s
{
    uv_buf_t buf;
    QUEUE queue_item;
    int index;
}hbuf_t, * p_hbuf_t;


//all memory operation is malloc & free  
#define CONNECTION_FIELDS                                                    \
   uv_tcp_t tcp;                                                             \
   net_object_type object_type;                                              \
   e_connection_state state;                                                 \
   connection_id cid;                                                        \
   struct connection_posted cp;                                         \
   phnet_ev_rv ev_rv;


#define CONNECTION_PRIVATE_FIELDS                                            \
   RB_ENTRY(connection) connection_node;

#define CONNECTOR_PRIVATE_FIELDS                                             \
   RB_ENTRY(connector) conector_node;                                        \
   uv_connect_t connect_req;                                                 \
   char name[MAX_NET_OBJECT_ID_LEN];                                         \
   uv_timer_t timer;                                                         \
   char ip[MAX_IP_LEN];                                                      \
   int port ;                                                                \
   int ip4true6false;                                                        \
   int reconnct_gap;                                                         \
   int active_close;

#define ACCEOTOR_PRIVATE_FIELDS                                              \
   RB_ENTRY(acceptor) acceptor_node;                                         \
   struct CONNECTION_META_RBT connection_meta_rbt;                           \
   char name[MAX_NET_OBJECT_ID_LEN];                                         \
   char ip[MAX_IP_LEN];                                                      \
   int ip4true6false;                                                        \
   int port;


#define OP_CONNECT_PRIVATE_FIELDS                                            \
char name[MAX_NET_OBJECT_ID_LEN];                                            \
char ip[MAX_IP_LEN];                                                         \
int port;                                                                    \
int ip4true6false;                                                           \
int reconnct_gap;                                                            \
connection_id conection_id;                                                  \
phnet_ev_rv ev_rv;


#define OP_ACCEPTOR_PRIVATE_FIELDS                                           \
char name[MAX_NET_OBJECT_ID_LEN];                                            \
char ip[MAX_IP_LEN];                                                         \
int port;                                                                    \
int ip4true6false;                                                           \
connection_id conection_id;                                                  \
phnet_ev_rv ev_rv;

#define OP_SEND_PRIVATE_FIELDS                                               \
connection_id conection_id;                                                  \
hbuf_t buf;                                                                  \
uv_write_t write_req;

#define OP_CLOSE_PRIVATE_FIELDS                                              \
connection_id conection_id;

#define OP_STOP_ACCEPTOR_PRIVATE_FIELDS                                      \
connection_id conection_id;

#define OP_STOP_CONNECTOR_PRIVATE_FIELDS                               \
connection_id conection_id;

#define OP_HB_PRIVATE_FIELDS                               

///这个非常重要 
struct connection_convert
{
    CONNECTION_FIELDS
};

struct connection
{
    CONNECTION_FIELDS
        CONNECTION_PRIVATE_FIELDS
};

struct connector
{
    CONNECTION_FIELDS
        CONNECTOR_PRIVATE_FIELDS
};

struct acceptor
{
    CONNECTION_FIELDS
        ACCEOTOR_PRIVATE_FIELDS
};

struct op_connect
{
    struct uv_app_s op;
    OP_CONNECT_PRIVATE_FIELDS
};

struct op_accept
{
    struct uv_app_s op;
    OP_ACCEPTOR_PRIVATE_FIELDS
};

struct op_send
{
    struct uv_app_s op;
    OP_SEND_PRIVATE_FIELDS
};

struct op_close
{
    struct uv_app_s op;
    OP_CLOSE_PRIVATE_FIELDS
};

struct op_stop_acceptor
{
    struct uv_app_s op;
    OP_STOP_ACCEPTOR_PRIVATE_FIELDS
};

struct op_stop_connector
{
    struct uv_app_s op;
    OP_STOP_CONNECTOR_PRIVATE_FIELDS
};

struct op_hb
{
    struct uv_app_s op;
    OP_HB_PRIVATE_FIELDS
};

struct op_stop_all
{
    struct uv_app_s op;
};

CMP_TI(connection_cmp, connection, cid)
RB_HEAD(CONNECTION_RBT, connection);
RB_GENERATE_STATIC(CONNECTION_RBT, connection, connection_node, connection_cmp)


CMP_TI(connector_cmp, connector, cid)
RB_HEAD(CONNECTOR_RBT, connector);
RB_GENERATE_STATIC(CONNECTOR_RBT, connector, conector_node, connector_cmp)

CMP_TI(acceptor_cmp, acceptor, cid)
RB_HEAD(ACCEPTOR_RBT, acceptor);
RB_GENERATE_STATIC(ACCEPTOR_RBT, acceptor, acceptor_node, acceptor_cmp)
static void timer_reconnect(uv_timer_t* handle);

struct hloop
{
    uv_loop_t loop;
    uv_thread_t thread_id;
    uv_timer_t guard_timer;
    int stop_flag;
    int connection_num;
    int acceptor_num;
    int connector_num;
    int index;
    uv_mutex_t mutex_loop_state;
    e_loop_state loop_state;
    struct CONNECTOR_RBT connector_rbt;
    struct ACCEPTOR_RBT acceptor_rbt;
    struct CONNECTION_RBT connections_rbt;
    struct ACCEPTOR_RBT acceptor_rbt_to_kill;
    struct connection_meta meta_find;
    uv_sem_t logic_sem;
    QUEUE events_queue;
    struct app_handle h;
};

void set_loop_state(struct hloop* lp, e_loop_state e) {
    uv_mutex_lock(&lp->mutex_loop_state);
    lp->loop_state = e;
    uv_mutex_unlock(&lp->mutex_loop_state);
}

e_loop_state get_loop_state(struct hloop* lp) {
    e_loop_state e;
    uv_mutex_lock(&lp->mutex_loop_state);
    e = lp->loop_state;
    uv_mutex_unlock(&lp->mutex_loop_state);
    return e;
}

struct connection_id_help
{
    connection_id id_mask;
    connection_id index[(1 << BIT_NUM)];
};

struct hnet
{
    struct hloop hoops[(1 << BIT_NUM)];
    struct connection_id_help cih;
    uv_mutex_t mutex_string_to_id;
    struct NET_OBJECT_STR2ID net_object_string_to_id_rbt;
    uv_key_t tls_connctor;
    uv_key_t tls_connection;
    uv_key_t tls_acceptor;
    uv_mutex_t mutex_net_event;
    QUEUE net_event_list;
    uv_mutex_t lb_accetpor;
    uv_mutex_t lb_connector;
    struct RB_APP_HANDLE handle;
};

static struct hnet hh_net;
static connection_id null_connection_id = 0;

struct net_object_string_to_id* get_net_obejct(const char* guid) {
    struct net_object_string_to_id null_find;
    strcpy_s(null_find.name, sizeof(null_find.name), guid);
    uv_mutex_lock(&hh_net.mutex_string_to_id);
    struct net_object_string_to_id* pfound = NET_OBJECT_STR2ID_RB_FIND(&hh_net.net_object_string_to_id_rbt, &null_find);
    uv_mutex_unlock(&hh_net.mutex_string_to_id);
    return pfound;
}

void remove_net_obejct(const char* guid) {
    struct net_object_string_to_id null_find;
    strcpy_s(null_find.name, sizeof(null_find.name), guid);
    uv_mutex_lock(&hh_net.mutex_string_to_id);
    struct net_object_string_to_id* pfound = NET_OBJECT_STR2ID_RB_FIND(&hh_net.net_object_string_to_id_rbt, &null_find);
    if (pfound != NULL) {
        NET_OBJECT_STR2ID_RB_REMOVE(&hh_net.net_object_string_to_id_rbt, pfound);
        free(pfound);
    }
    uv_mutex_unlock(&hh_net.mutex_string_to_id);
}

struct net_object_string_to_id* try_insert_net_object(const char* guid, connection_id cidd) {
    struct net_object_string_to_id* null_find = malloc(sizeof(struct net_object_string_to_id));
    ASSERT(NULL != null_find);
    memset(null_find, 0, sizeof(*null_find));
    strcpy_s(null_find->name, MAX_NET_OBJECT_ID_LEN, guid);
    null_find->id = cidd;
    uv_mutex_lock(&hh_net.mutex_string_to_id);
    struct net_object_string_to_id* p = NET_OBJECT_STR2ID_RB_FIND(&hh_net.net_object_string_to_id_rbt, null_find);
    if (p) {
        uv_mutex_unlock(&hh_net.mutex_string_to_id);
        free(null_find);
        HLOG_WARN("repeat acceptor name error %s\n", guid);
        return NULL;
    }
    NET_OBJECT_STR2ID_RB_INSERT(&hh_net.net_object_string_to_id_rbt, null_find);
    uv_mutex_unlock(&hh_net.mutex_string_to_id);
    return null_find;
}

struct op_connect* create_op_connect() {
    struct op_connect* p = malloc(sizeof(struct op_connect));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_connect));
    p->op.data = (void*)(OP_CONNECT);
    return p;
}

struct op_accept* create_op_accept() {
    struct op_accept* p = malloc(sizeof(struct op_accept));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_accept));
    p->op.data = (void*)(OP_ACCEPT);
    return p;
}

struct op_send* create_op_send() {
    struct op_send* p = malloc(sizeof(struct op_send));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_send));
    p->op.data = (void*)(OP_SEND);
    return p;
}

struct op_close* create_op_close() {
    struct op_close* p = malloc(sizeof(struct op_close));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_close));
    p->op.data = (void*)(OP_CLOSE);
    return p;
}

struct op_stop_acceptor* create_op_stop_acceptor() {
    struct op_stop_acceptor* p = malloc(sizeof(struct op_stop_acceptor));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_stop_acceptor));
    p->op.data = (void*)(OP_STOP_ACCEPTOR);
    return p;
}

struct op_stop_connector* create_op_stop_connector() {
    struct op_stop_connector* p = malloc(sizeof(struct op_stop_connector));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_stop_connector));
    p->op.data = (void*)(OP_STOP_CONNECTOR);
    return p;
}

struct op_stop_all* create_op_stop_all() {
    struct op_stop_all* p = malloc(sizeof(struct op_stop_all));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_stop_all));
    p->op.data = (void*)(OP_STOP_ALL);
    return p;
}

struct op_stop_all* create_op_hb() {
    struct op_hb* p = malloc(sizeof(struct op_hb));
    ASSERT(p != NULL);
    memset(p, 0, sizeof(struct op_hb));
    p->op.data = (void*)(OP_HB);
    return p;
}

net_app_on_connect_success* create_net_app_on_connect_success(phnet_ev_rv pevrv, int index, connection_id id) {
    net_app_on_connect_success* p = malloc(sizeof(net_app_on_connect_success));
    ASSERT(p != NULL);
    QUEUE_INIT(&(p->event_node));
    p->net_app_type = E_NET_APP_ON_CONNECT_SUCCESS;
    p->index = index;
    p->net_object_id = id;
    p->ev_rv = pevrv;
    return p;
}

net_app_on_connect_fail* create_net_app_on_connect_fail(phnet_ev_rv pevrv, int index, connection_id id) {
    net_app_on_connect_fail* p = malloc(sizeof(net_app_on_connect_fail));
    ASSERT(p != NULL);
    QUEUE_INIT(&(p->event_node));
    p->net_app_type = E_NET_APP_ON_CONNECT_FAIL;
    p->index = index;
    p->net_object_id = id;
    p->ev_rv = pevrv;
    return p;
}

net_app_on_connection* create_net_app_on_connection(phnet_ev_rv pevrv, int index, connection_id id) {
    net_app_on_connection* p = malloc(sizeof(net_app_on_connection));
    ASSERT(p != NULL);
    QUEUE_INIT(&(p->event_node));
    p->net_app_type = E_NET_APP_ON_CONNECTION;
    p->index = index;
    p->net_object_id = id;
    p->ev_rv = pevrv;
    return p;
}

net_app_on_receive* create_net_app_on_receive(phnet_ev_rv pevrv, int index, connection_id id) {
    net_app_on_receive* p = malloc(sizeof(net_app_on_receive));
    ASSERT(p != NULL);
    QUEUE_INIT(&(p->event_node));
    p->net_app_type = E_NET_APP_ON_RECEIVE;
    p->index = index;
    p->net_object_id = id;
    p->ev_rv = pevrv;
    return p;
}

net_app_on_close* create_net_app_on_close(phnet_ev_rv pevrv, int index, connection_id id) {
    net_app_on_close* p = malloc(sizeof(net_app_on_close));
    ASSERT(p != NULL);
    QUEUE_INIT(&(p->event_node));
    p->net_app_type = E_NET_APP_ON_CLOSE;
    p->index = index;
    p->net_object_id = id;
    p->ev_rv = pevrv;
    return p;
}

void free_net_app_on_connect_success(net_app_on_connect_success* p) {
    free(p);
}

void free_net_app_on_connect_fail(net_app_on_connect_fail* p) {
    free(p);
}

void free_net_app_on_connection(net_app_on_connection* p) {
    free(p);
}

void free_net_app_on_receive(net_app_on_receive* p) {
    free(p);
}

void free_net_app_on_close(net_app_on_close* p) {
    free(p);
}

//later on for memory pool use !
void free_op_send(struct op_send* p) {
    free(p->buf.buf.base);
    free(p);
}

void free_op_connect(struct op_connect* p) {
    free(p);
}

void free_op_accept(struct op_accept* p) {
    free(p);
}

void free_op_close(struct op_close* p) {
    free(p);
}

void free_op_stop_acceptor(struct op_stop_acceptor* p) {
    free(p);
}

void free_op_stop_connector(struct op_stop_connector* p) {
    free(p);
}

void free_op_stop_all(struct op_stop_all* p) {
    free(p);
}
void free_op_hb(struct op_hb* p) {
    free(p);
}
void free_connector(struct connector* p) {
    //HLOG_ERROR("free connector [%llu] [%p] \n", p->cid, p);
    free(p);
}

void free_acceptor(struct acceptor* p) {
    free(p);
}

void free_connection(struct connection* p) {
    //HLOG_ERROR("free connection [%llu] [%p] \n", p->cid, p);
    free(p);
}

void init_bit_op() {
    connection_id id = (1 << BIT_NUM) - 1;
    hh_net.cih.id_mask = id << ((sizeof(connection_id) * 8 - BIT_NUM));
    for (int i = 0; i < ((1 << BIT_NUM)); i++) {
        hh_net.cih.index[i] = 1;
    }
}

int is_valid_index(int idx) {
    if (idx < 0 && idx >= (1 << BIT_NUM)) {
        return 0;
    }
    return 1;
}

connection_id generate_connection_id(int idx) {
    if (!is_valid_index(idx)) {
        return null_connection_id;
    }
    connection_id id = (connection_id)(idx);
    id <<= ((sizeof(connection_id) * 8 - BIT_NUM));
    id += (hh_net.cih.index[idx]++);
    return id;
}

int get_connection_index(connection_id id) {
    id &= hh_net.cih.id_mask;
    id >>= ((sizeof(connection_id) * 8 - BIT_NUM));
    return (int)id;
}

struct hloop* get_hloop(connection_id cid) {
    int hid = get_connection_index(cid);
    if (!is_valid_index(hid)) {
        return NULL;
    }
    return &hh_net.hoops[hid];
}
struct hloop* get_hloop_from_index(int idx) {
    if (!is_valid_index(idx)) {
        return NULL;
    }
    return &hh_net.hoops[idx];
}

struct hloop* select_connector_hloop() {
    struct hloop* ret = NULL;
    int min_connector = MAX_INT32;
    uv_mutex_lock(&hh_net.lb_connector);
    for (int i = 0; i < ((1 << BIT_NUM)); i += 2) {
        int tmp = hh_net.hoops[i].connector_num;
        if (tmp < min_connector) {
            ret = &hh_net.hoops[i];
            min_connector = tmp;
        }
    }
    uv_mutex_unlock(&hh_net.lb_connector);
    return ret;
}

struct hloop* select_acceptor_hloop() {
    struct hloop* ret = NULL;
    int min_acceptor = MAX_INT32;
    uv_mutex_lock(&hh_net.lb_accetpor);
    for (int i = 1; i < ((1 << BIT_NUM)); i += 2) {
        int tmp = hh_net.hoops[i].acceptor_num;
        if (tmp < min_acceptor) {
            ret = &hh_net.hoops[i];
            min_acceptor = tmp;
        }
    }
    uv_mutex_unlock(&hh_net.lb_accetpor);
    return ret;
}

int increase_acceptor_num(int index) {

    if (!is_valid_index(index)) {
        HLOG_ERROR("increase_acceptor_num invalid index [%d]\n", index);
        return 0;
    }
    uv_mutex_lock(&hh_net.lb_accetpor);
    int tmp = hh_net.hoops[index].acceptor_num;
    tmp++;
    hh_net.hoops[index].acceptor_num = tmp;
    uv_mutex_unlock(&hh_net.lb_accetpor);
    return tmp;
}

int decrease_acceptor_num(int index) {

    if (!is_valid_index(index)) {
        HLOG_ERROR("decrease_acceptor_num invalid index [%d]\n", index);
        return 0;
    }
    uv_mutex_lock(&hh_net.lb_accetpor);
    int tmp = hh_net.hoops[index].acceptor_num;
    tmp--;
    if (tmp < 0) {
        tmp = 0;
    }
    hh_net.hoops[index].acceptor_num = tmp;
    uv_mutex_unlock(&hh_net.lb_accetpor);
    return tmp;
}

int increase_connector_num(int index) {

    if (!is_valid_index(index)) {
        HLOG_ERROR("increase_connector_num invalid index [%d]\n", index);
        return 0;
    }
    uv_mutex_lock(&hh_net.lb_connector);
    int tmp = hh_net.hoops[index].connector_num;
    tmp++;
    hh_net.hoops[index].connector_num = tmp;
    uv_mutex_unlock(&hh_net.lb_connector);
    return tmp;
}

int decrease_connector_num(int index) {

    if (!is_valid_index(index)) {
        HLOG_ERROR("decrease_connector_num invalid index [%d]\n", index);
        return 0;
    }
    uv_mutex_lock(&hh_net.lb_connector);
    int tmp = hh_net.hoops[index].connector_num;
    tmp--;
    if (tmp < 0) {
        tmp = 0;
    }
    hh_net.hoops[index].connector_num = tmp;
    uv_mutex_unlock(&hh_net.lb_connector);
    return tmp;
}


p_hbuf_t get_buf_from_index(p_hbuf_t p, int index, const char* data, int len) {
    p->buf.base = malloc(len);
    memcpy_s(p->buf.base, len, data, len);
    p->buf.len = len;
    p->index = index;
    QUEUE_INIT(&p->queue_item);
    return p;
}

static void loop_run(void* arg) {
    struct hloop* loop = (struct one_loop*)arg;
    void* pconnctor = malloc(sizeof(struct connector));
    uv_key_set(&hh_net.tls_connctor, pconnctor);
    void* pacceptor = malloc(sizeof(struct acceptor));
    uv_key_set(&hh_net.tls_acceptor, pacceptor);
    void* pconnection = malloc(sizeof(struct connection));
    uv_key_set(&hh_net.tls_connection, pconnection);
    while ((loop)->stop_flag == 0) {
        QUEUE_INIT(&loop->events_queue);
        int r = uv_run(&loop->loop, UV_RUN_ONCE);
        if (!QUEUE_EMPTY(&loop->events_queue)) {
            uv_mutex_lock(&hh_net.mutex_net_event);
            // QUEUE_MOVE(&loop->events_queue, &hh_net.net_event_list);
          /*  while (!QUEUE_EMPTY(&loop->events_queue)) {
                QUEUE* p = QUEUE_NEXT(&loop->events_queue);
                QUEUE_REMOVE(p);
                QUEUE_INIT(p);
                QUEUE_INSERT_TAIL(&hh_net.net_event_list, p);
            }*/
            QUEUE_ADD(&hh_net.net_event_list, &loop->events_queue);//原先函数调用错误了 调用成了QUEUE_MOVE
            /*QUEUE* qIter = NULL;
            unsigned long long cnt = 0;
            static unsigned long long loop_times = 0;
            loop_times++;
            QUEUE_FOREACH(qIter, &hh_net.net_event_list) {
                cnt++;
            }
            SYNC_PACKAGE_lOG("times[%llu] total packet[%llu]\n", loop_times, cnt);*/
            uv_mutex_unlock(&hh_net.mutex_net_event);
        }
    }
    free(pconnctor);
    free(pconnection);
    free(pacceptor);
    struct acceptor* pacc = NULL;
    RB_FOREACH(pacc, ACCEPTOR_RBT, &loop->acceptor_rbt_to_kill) {
        free_acceptor(pacc);
    }
    HLOG_WARN("loop [%d] stop\n", loop->index);
}

#define LOOP_SIZE(a) (sizeof(a) / sizeof((a)[0]))

static void on_alloc(uv_handle_t* handle,
                     size_t suggested_size,
                     uv_buf_t* buf) {
    buf->base = malloc(suggested_size);
    memset(buf->base, 0, suggested_size);
    buf->len = suggested_size;
}

static void on_close(uv_handle_t* peer) {
    ///还要从其中移除
    struct connection_convert* pconvert = (struct connection_convert*)(peer);
    struct hloop* phoop = get_hloop(pconvert->cid);
    net_app_on_close* p_on_close = create_net_app_on_close(pconvert->ev_rv, phoop->index, pconvert->cid);
    QUEUE_INSERT_TAIL(&phoop->events_queue, &p_on_close->event_node);

    switch (pconvert->object_type) {
    case NET_OBJECT_CONNECTOR4:
    case NET_OBJECT_CONNECTOR6:
    {///是否有重连定时器删除问题
        ///主动关闭的话不需要重连，被动需要尝试
        struct connector* p_real_connector = (struct connector*)(peer);
        struct connector* pfind = uv_key_get(&hh_net.tls_connctor);
        ASSERT(pfind != NULL);
        pfind->cid = pconvert->cid;
        strcpy_s(pfind->name, sizeof(pfind->name), p_real_connector->name);
        struct connector* con = CONNECTOR_RBT_RB_FIND(&phoop->connector_rbt, pfind);
        if (con) {
            if (con->active_close || con->reconnct_gap == 0) {
                HLOG_INFO("on_close connector [%llu] success\n", pfind->cid);
                CONNECTOR_RBT_RB_REMOVE(&phoop->connector_rbt, con);
                remove_net_obejct(con->name);
                free_connector(con);
            } else {
                con->cp.post_type |= get_post_mask(POST_TIMER_RECONNECTING);
                uv_timer_init(&phoop->loop, &con->timer);
                uv_timer_start(&con->timer, timer_reconnect, con->reconnct_gap, 0);
            }
        } else {
            HLOG_ERROR("on_close not find connector [%llu]\n ", pconvert->cid);
            return;
        }
    }
    break;
    case NET_OBJECT_CONNECTION:
    {
        struct connection* pfind = uv_key_get(&hh_net.tls_connection);
        ASSERT(pfind != NULL);
        pfind->cid = pconvert->cid;
        struct connection* con = CONNECTION_RBT_RB_FIND(&phoop->connections_rbt, pfind);
        if (con) {
            CONNECTION_RBT_RB_REMOVE(&phoop->connections_rbt, con);
            ;
            phoop->meta_find.cid = pconvert->cid;
            RB_SET(&phoop->meta_find, NULL, node);
            struct uv_tcp_s* p = (struct uv_tcp_s*)(peer->data);
            struct acceptor* pacceptor = (struct acceptor*)(p->data);

            struct connection_meta* pmeta = CONNECTION_META_RBT_RB_FIND(&pacceptor->connection_meta_rbt, &phoop->meta_find);
            if (pmeta) {
                CONNECTION_META_RBT_RB_REMOVE(&pacceptor->connection_meta_rbt, pmeta);
                free(pmeta);
            }

            if (pacceptor->state == STATE_CLOSED /*|| pacceptor->state == STATE_CLOSING*/) {

                if (RB_EMPTY(&pacceptor->connection_meta_rbt)) {///
                    ACCEPTOR_RBT_RB_REMOVE(&phoop->acceptor_rbt, pacceptor);
                    //acceptor_rbt_to_kill
                    ACCEPTOR_RBT_RB_INSERT(&phoop->acceptor_rbt_to_kill, pacceptor);
                    //free_acceptor(pacceptor); 
                    HLOG_INFO("on_close connection[%llu] acceptor closed remove acceptor \n", pfind->cid);
                }
            }
            HLOG_INFO("on_close connection from acceptor [%llu] success\n", pfind->cid);
            free_connection((struct connection*)peer);
        } else {
            HLOG_ERROR("on_close not find acceptors connection [%llu]\n ", pconvert->cid);
            return;
        }
    }break;
    case NET_OBJECT_ACCEPTOR4:
    case NET_OBJECT_ACCEPTOR6:
    {
        struct acceptor* pacceptor = (struct acceptor*)(peer->data);
        pacceptor->state = STATE_CLOSED;
        SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pacceptor->cid, "STATE_CLOSED");

        if (pacceptor->state == STATE_CLOSED) {
            if (RB_EMPTY(&pacceptor->connection_meta_rbt)) {///all connections closed
                ACCEPTOR_RBT_RB_REMOVE(&phoop->acceptor_rbt, pacceptor);
                ///不能在这里删除 要到最后删除 把acceptor放到所有结束再删除 
                ///线程技术再删除 以防关闭时序问题崩溃 不优雅
                ACCEPTOR_RBT_RB_INSERT(&phoop->acceptor_rbt_to_kill, pacceptor);
                //free_acceptor(pacceptor);
                HLOG_INFO("on_close acceptor [%llu] remove acceptor \n", pacceptor->cid);
            }
        }
    }
    break;
    default:
    break;
    }
    if (phoop->loop_state == LOOP_STATE_CLOSED) {
        if ((phoop->index % 2) == 0) {
            if (RB_EMPTY(&phoop->connector_rbt)) {
                HLOG_INFO("hloop[%d]logic stop post_sem on_close loop_state == LOOP_STATE_CLOSED connector empty post sem \n", phoop->index);
                uv_sem_post(&phoop->logic_sem);
            }
        } else {
            if (RB_EMPTY(&phoop->connections_rbt)) {
                HLOG_INFO("hloop[%d]logic stop post_sem on_close loop_state == LOOP_STATE_CLOSED acceptor&connections empty post sem \n", phoop->index);
                uv_sem_post(&phoop->logic_sem);
            }
        }
    }

}

static void on_shutdown(uv_shutdown_t* req, int status) {
    struct connection_convert* pc = (struct connection_convert*)req->handle;
    pc->cp.post_type &= ~get_post_mask(POST_SHUTDOWNING);
    pc->cp.post_type |= get_post_mask(POST_CLOSEING);
    SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pc->cid);
    if (!(req->handle->flags & UV_HANDLE_CLOSING)) {
        uv_close((uv_handle_t*)req->handle, on_close);
    }
    /*if (pc->state != STATE_CLOSING
        &&
        pc->state != STATE_CLOSED
        ) {
    }*/
    free(req);
}

static void on_read(uv_stream_t* handle,
                    ssize_t nread,
                    const uv_buf_t* buf) {
    struct connection_convert* pconv = (struct connection_convert*)(handle);
    if (nread < 0) {
        /* Error or EOF */
        //ASSERT(nread == UV_EOF);
        free(buf->base);
        if (pconv->state == STATE_CONNECTED) {
            SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconv->cid, "STATE_CLOSING");

            pconv->state = STATE_CLOSING;
            pconv->cp.post_type |= get_post_mask(POST_CLOSEING);
            pconv->cp.reason = CLOSING_READ_ERROR;
            SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pconv->cid);
            uv_close((uv_handle_t*)handle, on_close);///just close it 
        }
        return;
    }

    if (nread == 0) {
        /* Everything OK, but nothing read. */
        free(buf->base);
        return;
    }
    int index = get_connection_index(pconv->cid);
    if (!is_valid_index(index)) {
        HLOG_ERROR("on_read connection[%llu] invalid index[%d] \n", index, pconv->cid);
        return;
    }

    net_app_on_receive* p_on_recv = create_net_app_on_receive(pconv->ev_rv, index, pconv->cid);
    p_on_recv->buf = buf->base; //malloc(nread);//  buf->base;
    //memcpy_s(p_on_recv->buf, nread, buf->base, nread);
    p_on_recv->len = nread;
    HLOG_INFO("on_read from connection[%llu] nread [%d] len [%d] data[%s] \n", pconv->cid, nread, buf->len, buf->base);
    //free(buf->base);
    //buf->base = NULL;
    QUEUE_INSERT_TAIL(&hh_net.hoops[index].events_queue, &p_on_recv->event_node);
}

static void reconnect_free_connector(uv_handle_t* handle) {

    struct connector* p = (struct connector*)QUEUE_DATA(handle, struct connector, timer);
    if (p) {
        HLOG_WARN("reconnect_free_connector name [%s] cid [%llu]\n", p->name, p->cid);
        QUEUE_REMOVE(&p->tcp.handle_queue);
        //HLOG_ERROR("free connector [%llu] [%p]\n", p->cid, p);
        free(p);

    }
}

static void timer_reconnect(uv_timer_t* handle) {
    struct connector* p = (struct connector*)QUEUE_DATA(handle, struct connector, timer);
    ASSERT(p != NULL);

    struct hloop* ploop = get_hloop(p->cid);// &hh_net.hoops[0].loop;
    if (ploop == NULL) {
        return;
    }
    int r = 0;
    int free_me = 0;

    HLOG_WARN("timer_reconnect [%llu]name[%s]\n", p->cid, p->name);
    p->state = STATE_CONNECTING;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, p->cid, "STATE_CONNECTING");
    p->cp.post_type |= get_post_mask(POST_TIMER_RECONNECTING);

    if (p->ip4true6false > 0) {
        struct sockaddr_in addr;
        ASSERT(0 == uv_ip4_addr(p->ip, p->port, &addr));
        r = uv_tcp_init(&ploop->loop, &p->tcp);
        p->tcp.data = p;
        p->object_type = NET_OBJECT_CONNECTOR4;
        ASSERT(0 == r);
        r = uv_tcp_connect(&p->connect_req,
                           &p->tcp,
                           (const struct sockaddr*)&addr,
                           on_connect);
        if (r != 0) {
            const char* error_name = uv_err_name(r);
            HLOG_ERROR("timer_reconnect ip44 error [%s]\n", error_name);
            CONNECTOR_RBT_RB_REMOVE(&(ploop->connector_rbt), p);
            remove_net_obejct(p->name);
            free_me = 1;
        }
    } else {
        struct sockaddr_in6 addr6;
        ASSERT(0 == uv_ip6_addr(p->ip, p->port, &addr6));
        r = uv_tcp_init(&ploop->loop, &p->tcp);
        p->tcp.data = p;
        p->object_type = NET_OBJECT_CONNECTOR6;
        ASSERT(0 == r);
        r = uv_tcp_connect(&p->connect_req,
                           &p->tcp,
                           (const struct sockaddr*)&addr6,
                           on_connect);
        if (r != 0) {
            const char* error_name = uv_err_name(r);
            HLOG_ERROR("timer_reconnect ip66 error [%s]\n", error_name);
            CONNECTOR_RBT_RB_REMOVE(&(ploop->connector_rbt), p);
            remove_net_obejct(p->name);
            free_me = 1;
        }
    }
    if (free_me == 0) {
        SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, p->cid);
        uv_close((uv_handle_t*)handle, NULL);
    } else {
        SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, p->cid);
        uv_close((uv_handle_t*)handle, reconnect_free_connector);
    }
}

static void on_connect_error(uv_handle_t* peer) {
    struct connector* pconnector = (struct connector*)(QUEUE_DATA(peer, struct connector, tcp));
    ASSERT(pconnector != NULL);
    pconnector->state = STATE_CLOSED;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconnector->cid, "STATE_CONNECTED");

    HLOG_WARN("on_connect_error [%llu]name[%s] state[%d]active [%d]\n", pconnector->cid, pconnector->name, pconnector->state, pconnector->active_close);
    struct hloop* ploop = get_hloop(pconnector->cid);//&hh_net.hoops[0];
    if (ploop == NULL) {
        return;
    }
    if (pconnector->reconnct_gap == 0) {
        CONNECTOR_RBT_RB_REMOVE(&ploop->connector_rbt, pconnector);
        remove_net_obejct(pconnector->name);
        HLOG_WARN("on_connect_error free_connector [%llu]name[%s]\n", pconnector->cid, pconnector->name);
        QUEUE_REMOVE(&pconnector->tcp.handle_queue);
        //HLOG_ERROR("free connector [%llu] [%p]\n", pconnector->cid, pconnector);
        free_connector(pconnector);
    }

    uv_timer_init(&ploop->loop, &pconnector->timer);
    uv_timer_start(&pconnector->timer, timer_reconnect, pconnector->reconnct_gap, 0);
}

static void on_connection(uv_stream_t* server, int status) {
    uv_stream_t* stream;
    int r = 0;
    if (status != 0) {
        ///TODO: how to handle the memory !
        HLOG_ERROR("on_connection onnect error %s\n", uv_err_name(status));
        return;
    }
    ASSERT(status == 0);
    struct acceptor* pacce = (struct acceptor*)(server->data);
    ASSERT(pacce != NULL);
    if (pacce->state == STATE_CLOSING || pacce->state == STATE_CLOSED) {
        HLOG_ERROR("on_connection but acceptor is closing [%d]\n", pacce->state);
        return;
    }
    struct connection* pcon = malloc(sizeof(struct connection));
    ASSERT(pcon != NULL);
    memset(pcon, 0, sizeof(struct connection));
    RB_SET(pcon, NULL, connection_node);
    stream = &pcon->tcp;
    struct hloop* ploop = ((struct hloop*)server->loop);
    pcon->cid = generate_connection_id(ploop->index);
    //HLOG_ERROR("malloc connection id[%llu] [%p]\n", pcon->cid, pcon);
    pcon->ev_rv = pacce->ev_rv;
    CONNECTION_RBT_RB_INSERT(&ploop->connections_rbt, pcon);
    pcon->object_type = NET_OBJECT_CONNECTION;
    pcon->state = STATE_CONNECTED;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pcon->cid, "STATE_CONNECTED");

    struct connection_meta* meta = malloc(sizeof(struct connection_meta));
    RB_SET(meta, NULL, node);
    meta->cid = pcon->cid;
    CONNECTION_META_RBT_RB_INSERT(&pacce->connection_meta_rbt, meta);
    ASSERT(stream != NULL);
    r = uv_tcp_init(ploop, (uv_tcp_t*)stream);
    /* associate server with stream */
    stream->data = server;
    r = uv_accept(server, stream);
    ASSERT(r == 0);
    pcon->cp.post_type |= get_post_mask(POST_READING);
    r = uv_read_start(stream, on_alloc, on_read);
    ASSERT(r == 0);

    struct sockaddr sockname, peername;
    int namelen = sizeof sockname;
    r = uv_tcp_getsockname(stream, &sockname, &namelen);
    r = uv_tcp_getpeername(stream, &peername, &namelen);
    net_app_on_connection* p_on_connection = create_net_app_on_connection(pacce->ev_rv, ploop->index, pcon->cid);
    struct sockaddr_in local_addr = *(struct sockaddr_in*)&sockname;
    r = uv_ip4_name(&local_addr, (char*)p_on_connection->local_ip, sizeof p_on_connection->local_ip);
    ASSERT(r == 0);
    p_on_connection->local_port = ntohs(local_addr.sin_port);
    struct sockaddr_in remote_addr = *(struct sockaddr_in*)&peername;
    r = uv_ip4_name(&remote_addr, (char*)p_on_connection->remote_ip, sizeof p_on_connection->remote_ip);
    ASSERT(r == 0);
    p_on_connection->remote_port = ntohs(remote_addr.sin_port);
    strcpy_s(p_on_connection->name, sizeof p_on_connection->name, pacce->name);
    HLOG_INFO("on_connection id [%llu] remote ip [%s] port[%d] local ip [%s] port[%d]\n", pcon->cid, p_on_connection->remote_ip, p_on_connection->remote_port,
              p_on_connection->local_ip, p_on_connection->local_port);
    QUEUE_INSERT_TAIL(&ploop->events_queue, &p_on_connection->event_node);
}

static void on_connect(uv_connect_t* req, int status) {

    struct connector* pconnector = (struct connector*)req->handle->data;
    struct hloop* plooop = get_hloop(pconnector->cid);
    if (plooop == NULL) {
        return;
    }
    ///HLOG_WARN("on_connect_return_test [%s] [%llu] state[%d] active [%d]\n", pconnector->name, pconnector->cid, pconnector->state, pconnector->active_close);
    if (pconnector->active_close) {
        pconnector->cp.post_type |= get_post_mask(POST_CLOSEING);
        pconnector->cp.reason = CLOSING_ACTIVE;
        SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pconnector->cid);
        SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconnector->cid, "STATE_CLOSING");

        // uv_close((uv_handle_t*)req->handle, on_close);
        if (pconnector->state != STATE_CLOSING
            &&
            pconnector->state != STATE_CLOSED
            ) {
            pconnector->state = STATE_CLOSING;
            uv_close((uv_handle_t*)req->handle, on_close);
        }

        HLOG_WARN("on_connect but active closed [%s] \n", pconnector->name);
        return;
    }
    if (status != 0) {
        SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconnector->cid, "STATE_CLOSING");

        //pconnector->state = STATE_CLOSING;
        pconnector->cp.post_type |= get_post_mask(POST_CLOSEING);
        pconnector->cp.reason = CLOSING_CONNECT_ERROR;
        SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pconnector->cid);
        //uv_close((uv_handle_t*)req->handle, on_connect_error);
       /* if (pconnector->state != STATE_CLOSING
            &&
            pconnector->state != STATE_CLOSED
            ) {
            pconnector->state = STATE_CLOSING;
            uv_close((uv_handle_t*)req->handle, on_connect_error);
        }*/

        if (!(req->handle->flags & UV_HANDLE_CLOSING)) {
            pconnector->state = STATE_CLOSING;
            uv_close((uv_handle_t*)req->handle, on_connect_error);
        }

        net_app_on_connect_fail* p_on_connect_fail = create_net_app_on_connect_fail(pconnector->ev_rv, 0, pconnector->cid);
        strcpy_s(p_on_connect_fail->name, sizeof p_on_connect_fail->name, pconnector->name);

        QUEUE_INSERT_TAIL(&plooop->events_queue, &p_on_connect_fail->event_node);
    } else {
        SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconnector->cid, "STATE_CONNECTED");

        pconnector->state = STATE_CONNECTED;
        pconnector->cp.post_type |= get_post_mask(POST_READING);
        uv_read_start((uv_stream_t*)(req->handle), on_alloc, on_read);
        struct sockaddr sockname, peername;
        int namelen = sizeof sockname;
        int r = uv_tcp_getsockname(req->handle, &sockname, &namelen);
        r = uv_tcp_getpeername(req->handle, &peername, &namelen);
        net_app_on_connect_success* p_on_connect = create_net_app_on_connect_success(pconnector->ev_rv, 0, pconnector->cid);

        struct sockaddr_in local_addr = *(struct sockaddr_in*)&sockname;
        r = uv_ip4_name(&local_addr, (char*)p_on_connect->local_ip, sizeof p_on_connect->local_ip);
        ASSERT(r == 0);
        p_on_connect->local_port = ntohs(local_addr.sin_port);
        struct sockaddr_in remote_addr = *(struct sockaddr_in*)&peername;
        r = uv_ip4_name(&remote_addr, (char*)p_on_connect->remote_ip, sizeof p_on_connect->remote_ip);
        ASSERT(r == 0);
        p_on_connect->remote_port = ntohs(remote_addr.sin_port);
        strcpy_s(p_on_connect->name, sizeof p_on_connect->name, pconnector->name);
        QUEUE_INSERT_TAIL(&plooop->events_queue, &p_on_connect->event_node);
        HLOG_INFO("on_connect [%s] remote ip [%s] port[%d] local ip [%s] port[%d] handle  \n",
                  pconnector->name, p_on_connect->remote_ip, p_on_connect->remote_port,
                  p_on_connect->local_ip, p_on_connect->local_port);
    }
}

static void on_write(uv_write_t* req, int status) {

    struct op_send* p_op_send = QUEUE_DATA(req, struct op_send, write_req);

    HLOG_WARN("connection id [%llu]write error [%d]\n", p_op_send->conection_id, status);
    //only something wrong goes here 
    int index = get_connection_index(p_op_send->conection_id);
    if (is_valid_index(index) == 0) {
        HLOG_WARN("connection id [%llu]write error [%d] invalid index [%d]\n", p_op_send->conection_id, status, index);
        goto __r;
    }
    struct hloop* p_loop = &hh_net.hoops[index];
    if ((index % 2) == 0) {
        struct connector* tls_conector = uv_key_get(&hh_net.tls_connctor);
        if (!tls_conector) {
            HLOG_WARN("connection id [%llu]write error [%d] not find tls connector\n", p_op_send->conection_id, status);
            goto __r;
        }
        tls_conector->cid = p_op_send->conection_id;
        struct connector* pfind = CONNECTOR_RBT_RB_FIND(&p_loop->connector_rbt, tls_conector);
        if (!pfind) {
            HLOG_WARN("connection id [%llu]write error [%d] not find connector\n", p_op_send->conection_id, status);
            goto __r;
        }

        if (status == 0) {
            pfind->cp.write_times--;
            if (pfind->cp.write_times == 0) {
                pfind->cp.post_type &= ~get_post_mask(POST_WRITEING);
            }
            goto __r;
        }

        if (pfind->state == STATE_CONNECTED) {
            SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pfind->cid, "STATE_CLOSING");

            pfind->state = STATE_CLOSING;
            pfind->cp.post_type |= get_post_mask(POST_CLOSEING);
            pfind->cp.reason = CLOSING_WRITE_ERROR;
            SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pfind->cid);
            uv_close(&pfind->tcp, on_connect_error);
        }

    } else {
        struct connection* tls_connection = uv_key_get(&hh_net.tls_connection);
        if (!tls_connection) {
            HLOG_WARN("connection id [%llu]write error [%d] not find tls connection\n", p_op_send->conection_id, status);
            goto __r;
        }
        tls_connection->cid = p_op_send->conection_id;
        struct connection* pfind = CONNECTION_RBT_RB_FIND(&p_loop->connector_rbt, tls_connection);
        if (!pfind) {
            HLOG_WARN("connection id [%llu]write error [%d] not find connection\n", p_op_send->conection_id, status);
            goto __r;
        }
        if (status == 0) {
            pfind->cp.write_times--;
            if (pfind->cp.write_times == 0) {
                pfind->cp.post_type &= ~get_post_mask(POST_WRITEING);
            }
            goto __r;
        }
        if (pfind->state == STATE_CONNECTED) {
            SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pfind->cid, "STATE_CLOSING");

            pfind->state = STATE_CLOSING;
            pfind->cp.post_type |= get_post_mask(POST_CLOSEING);
            pfind->cp.reason = CLOSING_WRITE_ERROR;
            SYNC_CLOSE_lOG("uv_close [%s,%d ,%llu]\n", __FILE__, __LINE__, pfind->cid);
            uv_close(&pfind->tcp, on_connect_error);
        }
    }
__r:
    free_op_send(p_op_send);
    return;
}

static void handle_op_connect(uv_loop_t* ploop, struct uv_app_s* ptr) {
    struct op_connect* p = (struct op_connect*)(ptr);
    struct hloop* phloop = (struct hloop*)(ploop);
    ASSERT(phloop != NULL);
    struct connector* pconnector = malloc(sizeof(struct connector));
    ASSERT(pconnector != NULL);
    memset(pconnector, 0, sizeof(struct connector));
    RB_SET(pconnector, NULL, conector_node);
    strcpy_s(pconnector->name, sizeof(pconnector->name), p->name);
    pconnector->cid = p->conection_id;
    pconnector->ip4true6false = p->ip4true6false;
    pconnector->reconnct_gap = p->reconnct_gap;
    strcpy_s(pconnector->ip, sizeof(pconnector->ip), p->ip);
    pconnector->port = p->port;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pconnector->cid, "STATE_CONNECTING");

    pconnector->state = STATE_CONNECTING;
    pconnector->ev_rv = p->ev_rv;
    //HLOG_ERROR("malloc connector[%llu][%p]\n", pconnector->cid, pconnector);
    CONNECTOR_RBT_RB_INSERT(&phloop->connector_rbt, pconnector);
    int r = 0;
    if (p->ip4true6false > 0) {
        struct sockaddr_in addr;
        ASSERT(0 == uv_ip4_addr(p->ip, p->port, &addr));
        r = uv_tcp_init(ploop, &pconnector->tcp);
        pconnector->tcp.data = pconnector;
        pconnector->object_type = NET_OBJECT_CONNECTOR4;
        ASSERT(0 == r);
        //HLOG_INFO("connector_handle_addr[%p][%s][%llu]\n", &pconnector->tcp.handle_queue, pconnector->name, pconnector->cid);
        pconnector->cp.post_type |= get_post_mask(POST_CONNECTING);
        r = uv_tcp_connect(&pconnector->connect_req,
                           &pconnector->tcp,
                           (const struct sockaddr*)&addr,
                           on_connect);
        if (r != 0) {
            const char* error_name = uv_err_name(r);
            HLOG_ERROR("connect ip4 error [%s]\n", error_name);
            CONNECTOR_RBT_RB_REMOVE(&phloop->connector_rbt, pconnector);
            remove_net_obejct(pconnector->name);
            QUEUE_REMOVE(&pconnector->tcp.handle_queue);
            //HLOG_ERROR("free connector[%llu][%p]\n", pconnector->cid, pconnector);
            free(pconnector);

        }
        // ASSERT(0 == r);
    } else {
        struct sockaddr_in6 addr6;
        ASSERT(0 == uv_ip6_addr(p->ip, p->port, &addr6));
        r = uv_tcp_init(ploop, &pconnector->tcp);
        pconnector->tcp.data = pconnector;
        pconnector->object_type = NET_OBJECT_CONNECTOR6;
        ASSERT(0 == r);
        pconnector->cp.post_type |= get_post_mask(POST_CONNECTING);
        r = uv_tcp_connect(&pconnector->connect_req,
                           &pconnector->tcp,
                           (const struct sockaddr*)&addr6,
                           on_connect);
        if (r != 0) {
            const char* error_name = uv_err_name(r);
            HLOG_ERROR("connect ip6 error [%s]\n", error_name);
            CONNECTOR_RBT_RB_REMOVE(&phloop->connector_rbt, pconnector);
            remove_net_obejct(pconnector->name);
            QUEUE_REMOVE(&pconnector->tcp.handle_queue);
            free(pconnector);
        }
    }
}

static void handle_op_accept(uv_loop_t* ploop, struct uv_app_s* ptr) {
    struct op_accept* p = (struct op_accept*)(ptr);
    struct hloop* phloop = (struct hloop*)(ploop);
    ASSERT(phloop != NULL);
    struct acceptor* acceptor_find = uv_key_get(&hh_net.tls_acceptor);
    ASSERT(acceptor_find != NULL);
    acceptor_find->cid = p->conection_id;
    struct acceptor* pa_find = ACCEPTOR_RBT_RB_FIND(&phloop->acceptor_rbt, acceptor_find);
    //ASSERT(pa_find == NULL);
    if (pa_find != NULL) {
        HLOG_ERROR("op_accept same connection id [%llu]", p->conection_id);
        return;
    }
    struct acceptor* pa = malloc(sizeof(struct acceptor));
    ASSERT(pa != NULL);
    memset(pa, 0, sizeof * pa);
    RB_INIT(&pa->connection_meta_rbt);
    RB_SET(pa, NULL, acceptor_node);
    pa->state = STATE_CONNECTED;
    pa->cid = p->conection_id;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pa->cid, "STATE_CONNECTED");

    strcpy_s(pa->name, sizeof(pa->name), p->name);
    pa->ev_rv = p->ev_rv;
    // NET_OBJECT_STR2ID_RB_FIND
    ACCEPTOR_RBT_RB_INSERT(&phloop->acceptor_rbt, pa);
    int r = 0;
    if (p->ip4true6false > 0) {
        struct sockaddr_in addr;
        ASSERT(0 == uv_ip4_addr(p->ip, p->port, &addr));
        pa->cid = p->conection_id;
        r = uv_tcp_init(ploop, &pa->tcp);
        pa->tcp.data = pa;
        pa->object_type = NET_OBJECT_ACCEPTOR4;
        ASSERT(0 == r);
        r = uv_tcp_bind(&pa->tcp, (const struct sockaddr*)&addr, 0);
        ASSERT(0 == r);
        r = uv_listen((uv_stream_t*)&pa->tcp, SOMAXCONN, on_connection);
        if (r) {
            HLOG_ERROR("listen error %s\n", uv_err_name(r));
        }
    } else {
        struct sockaddr_in6 addr6;
        ASSERT(0 == uv_ip6_addr(p->ip, p->port, &addr6));
        pa->cid = p->conection_id;
        r = uv_tcp_init(ploop, &pa->tcp);
        pa->tcp.data = pa;
        pa->object_type = NET_OBJECT_ACCEPTOR6;
        if (r) {
            HLOG_ERROR("socket creation error\n");
        }
        r = uv_tcp_bind(&pa->tcp, (const struct sockaddr*)&addr6, 0);
        if (r) {
            HLOG_ERROR("ipv6 not supported\n");
        }
        r = uv_listen((uv_stream_t*)&pa->tcp, SOMAXCONN, on_connection);
        if (r) {
            HLOG_ERROR("listen error\n");
        }
    }
}

static void handle_op_send(uv_loop_t* loop, struct uv_app_s* ptr) {
    struct op_send* p = (struct op_send*)(ptr);
    struct hloop* phloop = (struct hloop*)(loop);
    ASSERT(phloop != NULL);
    if ((phloop->index % 2) == 0) {
        struct connector* connector_find = uv_key_get(&hh_net.tls_connctor);
        ASSERT(connector_find != NULL);
        connector_find->cid = p->conection_id;
        struct connector* pfind_connector = CONNECTOR_RBT_RB_FIND(&phloop->connector_rbt, connector_find);
        if (!pfind_connector) {
            HLOG_WARN("send data could not find conector [%llu]\n", p->conection_id);
            return;
        }
        if (pfind_connector->state != STATE_CONNECTED) {
            HLOG_WARN("send data conector [%llu] state wrong[%d] \n", pfind_connector->cid, pfind_connector->state);
            return;
        }
        connector_find->cp.post_type |= get_post_mask(POST_WRITEING);
        connector_find->cp.write_times++;
        if (uv_write(&p->write_req, &pfind_connector->tcp, &p->buf.buf, 1, on_write)) {
            HLOG_ERROR("conector uv_write failed [%llu]\n", p->conection_id);
        }
    } else {
        struct connection* connection_find = uv_key_get(&hh_net.tls_connection);
        ASSERT(connection_find != NULL);
        connection_find->cid = p->conection_id;
        struct connection* pfind_connection = CONNECTION_RBT_RB_FIND(&phloop->connections_rbt, connection_find);
        if (!pfind_connection) {
            HLOG_WARN("send data could not find connection [%llu]\n", p->conection_id);
            return;
        }
        if (pfind_connection->state != STATE_CONNECTED) {
            HLOG_WARN("send data conector [%llu] state wrong[%d]\n", pfind_connection->cid, pfind_connection->state);
            return;
        }
        pfind_connection->cp.post_type |= get_post_mask(POST_WRITEING);
        pfind_connection->cp.write_times++;
        if (uv_write(&p->write_req, &pfind_connection->tcp, &p->buf.buf, 1, on_write)) {
            HLOG_ERROR("connection uv_write failed [%llu]\n", p->conection_id);
        }
    }
}

static void handle_op_close(uv_loop_t* ploop, struct uv_app_s* ptr) {
    struct op_close* p = (struct op_close*)(ptr);
    struct hloop* phloop = (struct hloop*)(ploop);
    struct connection* connection_find = uv_key_get(&hh_net.tls_connection);
    ASSERT(connection_find != NULL);
    connection_find->cid = p->conection_id;
    struct connection* pfind_connection = CONNECTION_RBT_RB_FIND(&phloop->connections_rbt, connection_find);
    if (!pfind_connection) {
        HLOG_WARN("handle_op_close could not find connection [%llu]\n", p->conection_id);
        return;
    }
    if (pfind_connection->state != STATE_CONNECTED) {
        HLOG_WARN("handle_op_closea conector [%llu] state wrong[%d] \n", pfind_connection->cid, pfind_connection->state);
        return;
    }
    pfind_connection->cp.post_type |= get_post_mask(POST_SHUTDOWNING);
    pfind_connection->cp.reason = CLOSING_ACTIVE;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pfind_connection->cid, "STATE_CLOSING");

    pfind_connection->state = STATE_CLOSING;

    struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
    ASSERT(pshutdown != NULL);
    SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, pfind_connection->cid);

    uv_shutdown(pshutdown, (uv_stream_t*)&pfind_connection->tcp, on_shutdown);
}

static void handle_op_stop_connector(uv_loop_t* ploop, struct uv_app_s* p) {
    struct op_stop_connector* ptr = (struct op_stop_connector*)(p);
    struct hloop* phloop = (struct hloop*)(ploop);
    struct connector* pcon = uv_key_get(&hh_net.tls_connctor);
    ASSERT(pcon != NULL);
    pcon->cid = ptr->conection_id;
    struct connector* pfind_connector = CONNECTOR_RBT_RB_FIND(&phloop->connector_rbt, pcon);
    if (!pfind_connector) {
        HLOG_WARN("handle_op_stop_connector could not find connector[%llu]\n", ptr->conection_id);
        return;
    }
    pfind_connector->active_close = 1;
    if (pfind_connector->state != STATE_CONNECTED) {
        HLOG_WARN("handle_op_stop_connector connector[%llu] is not connected still\n", pcon->state, ptr->conection_id);
        return;
    }
    struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
    ASSERT(pshutdown != NULL);
    pcon->state = STATE_CLOSING;
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pcon->cid, "STATE_CLOSING");

    pcon->cp.post_type |= (get_post_mask(POST_SHUTDOWNING));
    pcon->cp.reason = CLOSING_ACTIVE;
    SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, pfind_connector->cid);

    uv_shutdown(pshutdown, (uv_stream_t*)&pfind_connector->tcp, on_shutdown);
}

static void handle_op_stop_acceptor(uv_loop_t* ploop, struct uv_app_s* p) {
    struct op_stop_acceptor* ptr = (struct op_stop_acceptor*)(p);

    struct hloop* phloop = (struct hloop*)(ploop);
    struct acceptor* paccetor = uv_key_get(&hh_net.tls_acceptor);
    ASSERT(paccetor != NULL);
    paccetor->cid = ptr->conection_id;
    struct acceptor* pfind_acceptor = ACCEPTOR_RBT_RB_FIND(&phloop->acceptor_rbt, paccetor);
    if (!pfind_acceptor) {
        HLOG_WARN("handle_op_stop_acceptor could not find acceptor[%llu]\n", ptr->conection_id);
        return;
    }
    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pfind_acceptor->cid, "STATE_CLOSING");

    pfind_acceptor->state = STATE_CLOSING;
    struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
    ASSERT(pshutdown != NULL);

    decrease_acceptor_num(phloop->index);
    struct connection* tls_find = uv_key_get(&hh_net.tls_connection);
    ASSERT(tls_find != NULL);
    struct connection_meta* tmp_con = NULL;
    RB_FOREACH(tmp_con, CONNECTION_META_RBT, &pfind_acceptor->connection_meta_rbt) {
        tls_find->cid = tmp_con->cid;
        struct connection* pcon = CONNECTION_RBT_RB_FIND(&phloop->connections_rbt, tls_find);
        if (pcon && (pcon->state != STATE_CLOSED && pcon->state != STATE_CLOSING)) {
            pcon->state = STATE_CLOSING;
            SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, pcon->cid, "STATE_CLOSING");

            struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
            ASSERT(pshutdown != NULL);
            pcon->cp.post_type |= (get_post_mask(POST_CLOSEING));
            pcon->cp.reason = CLOSING_ACTIVE;
            SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, pcon->cid);
            uv_shutdown(pshutdown, (uv_stream_t*)&pcon->tcp, on_shutdown);
        }
    }
    SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, pfind_acceptor->cid);

    uv_shutdown(pshutdown, (uv_stream_t*)&pfind_acceptor->tcp, on_shutdown);
    remove_net_obejct(pfind_acceptor->name);
}

static void handle_op_stop_all(uv_loop_t* ploop, struct uv_app_s* p) {
    struct op_stop_all* ptr = (struct op_stop_all*)(p);
    //do nothing here
    struct hloop* phloop = (struct hloop*)(ploop);
    phloop->loop_state = LOOP_STATE_CLOSED;
    if ((phloop->index % 2) == 0) {
        if (RB_EMPTY(&phloop->connector_rbt)) {
            HLOG_WARN("hloop[%d]logic stop post_sem\n", phloop->index);
            uv_sem_post(&phloop->logic_sem);
        } else {
            struct connector* tmp_con = NULL;
            RB_FOREACH(tmp_con, CONNECTOR_RBT, &phloop->connector_rbt) {
                tmp_con->active_close = 1;

                if (tmp_con->state == STATE_CONNECTED /*|| tmp_con->state == STATE_CONNECTING*/) {
                    HLOG_WARN("handle_op_stop_all hloop[%d]logic stop connector [%s] state ok state[%d]\n", phloop->index, tmp_con->name, tmp_con->state);
                    struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
                    ASSERT(pshutdown != NULL);
                    SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, tmp_con->cid);
                    uv_shutdown(pshutdown, (uv_stream_t*)&tmp_con->tcp, on_shutdown);
                    SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, tmp_con->cid, "STATE_CLOSING");

                    tmp_con->state = STATE_CLOSING;
                    tmp_con->cp.post_type |= (get_post_mask(POST_CLOSEING));
                    tmp_con->cp.reason = CLOSING_ACTIVE;
                } else {
                    HLOG_WARN("hloop[%d]logic stop connector [%s]state[%d] not ok\n", phloop->index, tmp_con->name, tmp_con->state);
                }
            }
        }
    } else {
        if (RB_EMPTY(&phloop->connections_rbt)) {
            //only think about the connections
            HLOG_WARN("hloop[%d]logic stop post_sem\n", phloop->index);
            uv_sem_post(&phloop->logic_sem);
        } else {
            do {

                struct connection* tmp_con = NULL;
                RB_FOREACH(tmp_con, CONNECTION_RBT, &phloop->connections_rbt) {
                    HLOG_WARN("hloop[%d]logic stop connection [%llu]\n", phloop->index, tmp_con->cid);
                    if (tmp_con->state != STATE_CLOSED && tmp_con->state != STATE_CLOSING) {
                        struct uv_shutdown_s* pshutdown = (struct uv_shutdown_s*)malloc(sizeof(struct uv_shutdown_s));
                        ASSERT(pshutdown != NULL);
                        SYNC_SET_STATE_lOG("uv_state [%s,%d ,%llu ,%s]\n", __FILE__, __LINE__, tmp_con->cid, "STATE_CLOSING");

                        tmp_con->state = STATE_CLOSING;
                        tmp_con->cp.post_type |= (get_post_mask(POST_CLOSEING));
                        tmp_con->cp.reason = CLOSING_ACTIVE;
                        SYNC_SHUTDOWN_lOG("uv_shutdown [%s,%d ,%llu]\n", __FILE__, __LINE__, tmp_con->cid);
                        uv_shutdown(pshutdown, (uv_stream_t*)&tmp_con->tcp, on_shutdown);
                    }
                }
            } while (0);
        }
    }
    free_op_stop_all(ptr);
}

static void handle_op_hb(uv_loop_t* ploop, struct uv_app_s* p) {
    struct op_hb* ptr = (struct op_hb*)(p);
    struct hloop* phloop = (struct hloop*)(ploop);
    hlog("heart_beat", LOG_WARNING, 1, "loop[%d] hb \n", phloop->index);
    free_op_hb(ptr);
}

struct app_handle* create_app_handle(e_op_type e) {
    switch (e) {
#define XX(uc,lc) case OP_##uc : { struct app_handle * h = malloc(sizeof (struct app_handle)) ;\
ASSERT(h!=NULL) ;\
        memset(h, 0, sizeof(struct app_handle));\
h->type = e ;\
        h->handle = lc;\
    RB_SET(h, NULL, node);\
return h ;\
        }
        APP_HANDLE_MAP(XX)
#undef XX
    default:
    break;
    }
    return NULL;
}

void init_app_handle() {
#define XX(uc,lc) do {\
    struct app_handle *h = create_app_handle(OP_##uc ) ;\
ASSERT(h!= NULL) ;\
    RB_APP_HANDLE_RB_INSERT(&hh_net.handle, h);\
    } while (0);
    APP_HANDLE_MAP(XX)
#undef XX 
}

void uninit_app_handle() {
    struct app_handle h;
#define XX(uc,lc) do {\
memset(&h,0,sizeof(struct app_handle)) ;\
RB_SET(&h,NULL,node);\
    struct app_handle* find = RB_APP_HANDLE_RB_FIND(&hh_net.handle, &h);\
        if (find != NULL) {\
                RB_APP_HANDLE_RB_REMOVE(&hh_net.handle, find);\
                free(find);\
        }\
    } while (0);
    APP_HANDLE_MAP(XX)
#undef XX
}

static void free_app_req(struct uv_app_s* ptr) {
    unsigned long long type = (unsigned long long)(ptr->data);
    switch (type) {
    case OP_CONNECT:
    case OP_ACCEPT:
    case OP_CLOSE:
    case OP_STOP_CONNECTOR:
    case OP_STOP_ACCEPTOR:
    {
        free(ptr);
    }
    break;
    case OP_SEND:
    case OP_STOP_ALL:
    {
        /* struct op_send* ptr1 = (struct op_send*)(ptr);
         free(ptr1->buf.buf.base);*/
    }  break;
    default: break;
    }

}

static void on_app_req(struct uv_app_s* ptr) {
    unsigned long long type = (unsigned long long)(ptr->data);
    struct hloop* phloop = (struct hloop*)(ptr->loop);
    HLOG_INFO("hloop[%d] receive command [%s]\n", phloop->index, get_op_name((e_op_type)type));
    RB_SET(&phloop->h, NULL, node);
    phloop->h.type = (int)(type);
    struct app_handle* ah = RB_APP_HANDLE_RB_FIND(&hh_net.handle, &phloop->h);
    if (ah) {
        ah->handle(ptr->loop, ptr);
    } else {
        HLOG_ERROR(__FUNCTION__ " faile to find handle for [%d]\n", (int)(type));
    }
    free_app_req(ptr);
}

static void guard_timer_cb(uv_timer_t* handle) {
    ASSERT(handle != NULL);
    ASSERT(1 == uv_is_active((uv_handle_t*)handle));
}
//暂时不考虑应用层多线程初始化和反初始化的状况
static int s_hnet_status = 0;
void init_hnet(int sbuf, int rbuf) {
    if (s_hnet_status != 0) {
        return;
    }
    s_hnet_status = 1;
    init_hlog(LOG_DEBUG, "logs");
    uv_set_send_buff_size(sbuf);
    uv_set_send_buff_size(rbuf);
    /*  if (c.on_connection == NULL ||
          c.on_connection == NULL ||
          c.on_receive == NULL ||
          c.on_close == NULL
          ) {
          HLOG_ERROR("init_hnet failed for hnet_callback's member has null pointer\n");
          abort();
          return;
      }*/
    memset(hh_net.hoops, 0, sizeof(hh_net.hoops));
    RB_INIT(&hh_net.net_object_string_to_id_rbt);
    //hh_net.cb = c;
    uv_mutex_init(&hh_net.mutex_string_to_id);
    ASSERT(0 == uv_key_create(&hh_net.tls_connctor));
    ASSERT(0 == uv_key_create(&hh_net.tls_connection));
    ASSERT(0 == uv_key_create(&hh_net.tls_acceptor));
    init_bit_op();
    uv_mutex_init(&hh_net.mutex_net_event);
    QUEUE_INIT(&hh_net.net_event_list);
    uv_mutex_init(&hh_net.lb_accetpor);
    uv_mutex_init(&hh_net.lb_connector);

    RB_INIT(&hh_net.handle);
    init_app_handle();
    int r = 0;
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        ///初始化 
        RB_INIT(&hh_net.hoops[i].connector_rbt);
        RB_INIT(&hh_net.hoops[i].acceptor_rbt);
        RB_INIT(&hh_net.hoops[i].acceptor_rbt_to_kill);
        hh_net.hoops[i].connection_num = 0;
        hh_net.hoops[i].acceptor_num = 0;
        hh_net.hoops[i].connector_num = 0;
        hh_net.hoops[i].stop_flag = 0;
        hh_net.hoops[i].index = i;
        uv_mutex_init(&hh_net.hoops[i].mutex_loop_state);
        ASSERT(0 == uv_sem_init(&hh_net.hoops[i].logic_sem, 0));
        ASSERT(0 == uv_loop_init(&hh_net.hoops[i].loop, on_app_req));
        uv_timer_init(&hh_net.hoops[i].loop, &hh_net.hoops[i].guard_timer);
        r = uv_timer_start(&hh_net.hoops[i].guard_timer, guard_timer_cb, 2000, 1);/* GUARD_TIMEER_GAP_MS, MAX_UINT64);*/
        ASSERT(r == 0);
        r = uv_thread_create(&hh_net.hoops[i].thread_id, loop_run, &hh_net.hoops[i]);
        ASSERT(r == 0);
    }
}

void uninit_hnet() {
    if (s_hnet_status != 1) {
        return;
    }
    s_hnet_status = 2;
    int r = 0;
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        if (i % 2 == 1) {
            continue;
        }
        struct op_stop_all* p_stop_all = create_op_stop_all();
        ASSERT(p_stop_all != NULL);
        p_stop_all->op.loop = &hh_net.hoops[i].loop;
        post_app_req(&hh_net.hoops[i].loop, p_stop_all);
        ASSERT(r == 0);
    }
    HLOG_ERROR("exit stage 1/8\n");
    //uv_sem_wait(&hh_net.hoops[0].logic_sem);
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        if (i % 2 == 1) {
            continue;
        }
        uv_sem_wait(&hh_net.hoops[i].logic_sem);
    }
    HLOG_ERROR("exit stage 2/8\n");
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        if (i % 2 != 1) {
            continue;
        }
        struct op_stop_all* p_stop_all = create_op_stop_all();
        ASSERT(p_stop_all != NULL);
        p_stop_all->op.loop = &hh_net.hoops[i].loop;
        post_app_req(&hh_net.hoops[i].loop, p_stop_all);
        ASSERT(r == 0);
    }
    HLOG_ERROR("exit stage 3/8\n");
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        if (i % 2 != 1) {
            continue;
        }
        uv_sem_wait(&hh_net.hoops[i].logic_sem);
    }
    HLOG_ERROR("exit stage 4/8\n");
    //关闭loop本身
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        hh_net.hoops[i].stop_flag = 1;
        Sleep(100);
        r = uv_timer_stop(&hh_net.hoops[i].guard_timer);
        uv_stop(&hh_net.hoops[i].loop);
        uv_loop_close(&hh_net.hoops[i].loop);
    }
    HLOG_ERROR("exit stage 5/8\n");
    //等待线程函数结束
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        r = uv_thread_join(&hh_net.hoops[i].thread_id);
    }
    HLOG_ERROR("exit stage 6/8\n");
    //内存的释放
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        uv_mutex_destroy(&hh_net.hoops[i].mutex_loop_state);
        uv_sem_destroy(&hh_net.hoops[i].logic_sem);
    }
    HLOG_ERROR("exit stage 7/8\n");
    //释放全局句柄
    uv_key_delete(&hh_net.tls_acceptor);
    uv_key_delete(&hh_net.tls_connctor);
    uv_key_delete(&hh_net.tls_connection);
    uv_mutex_destroy(&hh_net.mutex_string_to_id);
    uv_mutex_destroy(&hh_net.mutex_net_event);
    uv_mutex_destroy(&hh_net.lb_accetpor);
    uv_mutex_destroy(&hh_net.lb_connector);
    uninit_app_handle();
    HLOG_ERROR("exit stage 8/8 finally \n");
    s_hnet_status = 3;
    printf("bye!\n");
}

connection_id add_connector(phnet_ev_rv pcb, const char* cid, const char* ip, int port, int b4true6false, int reconnect_gap) {

    struct hloop* ploop = select_connector_hloop();
    connection_id cidd = generate_connection_id(ploop->index);
    ASSERT(NULL != ploop);
    struct net_object_string_to_id* try_insert = try_insert_net_object(cid, cidd);
    if (try_insert == NULL) {
        return null_connection_id;
    }
    increase_connector_num(ploop->index);

    struct op_connect* op_con = create_op_connect();
    UV_REQ_INIT(&op_con->op, UV_APP);
    op_con->op.loop = &ploop->loop;
    strcpy_s(op_con->name, sizeof op_con->name, cid);
    strcpy_s(op_con->ip, sizeof op_con->ip, ip);
    op_con->port = port;
    op_con->ip4true6false = b4true6false;
    op_con->reconnct_gap = reconnect_gap;
    op_con->conection_id = cidd;
    op_con->ev_rv = pcb;
    post_app_req(&ploop->loop, op_con);
    return cidd;
}

connection_id add_acceptor(phnet_ev_rv pcb, const char* aid, const char* ip, int port, int b4true6false) {
    struct hloop* ploop = select_acceptor_hloop();
    connection_id cidd = generate_connection_id(ploop->index);
    ASSERT(NULL != ploop);
    struct net_object_string_to_id* try_add = try_insert_net_object(aid, cidd);
    if (try_add == NULL) {
        return null_connection_id;
    }
    increase_acceptor_num(ploop->index);
    struct op_accept* op_ac = create_op_accept();
    UV_REQ_INIT(&op_ac->op, UV_APP);
    op_ac->op.loop = &ploop->loop;
    strcpy_s(op_ac->name, sizeof op_ac->name, aid);
    strcpy_s(op_ac->ip, sizeof op_ac->ip, ip);
    op_ac->port = port;
    op_ac->ip4true6false = b4true6false;
    op_ac->conection_id = cidd;
    op_ac->ev_rv = pcb;
    post_app_req(&ploop->loop, op_ac);
    return cidd;
}

void send_data(connection_id cid, const char* pdata, int len) {
    int index = get_connection_index(cid);
    if (!is_valid_index(index)) {
        HLOG_ERROR("send data invalid index [%d] connection id [%llu]\n", index, cid);
        return;
    }
    ///HLOG_INFO("send data to connection id [%llu] len [%d]\n", cid, len);

    struct hloop* ploop = get_hloop_from_index(index);
    ASSERT(ploop != NULL);
    struct op_send* op_send = create_op_send();
    UV_REQ_INIT(&op_send->op, UV_APP);
    op_send->op.loop = &ploop->loop;
    op_send->conection_id = cid;
    get_buf_from_index(&op_send->buf, index, pdata, len);
    post_app_req(&ploop->loop, op_send);
    return;
}

void close_connection(connection_id cid) {
    int index = get_connection_index(cid);
    if (!is_valid_index(index)) {
        HLOG_ERROR("close_connection invalid index [%d] connection id [%llu]\n", index, cid);
        return;
    }
    if ((index % 2) != 0) {
        struct hloop* ploop = get_hloop_from_index(index);
        ASSERT(ploop != NULL);
        struct op_close* op_close = create_op_close();
        UV_REQ_INIT(&op_close->op, UV_APP);
        op_close->op.loop = &ploop->loop;
        op_close->conection_id = cid;
        post_app_req(&ploop->loop, op_close);
    } else {
        struct op_stop_connector* p_op_stop_connector = create_op_stop_connector();
        ASSERT(p_op_stop_connector != NULL);
        p_op_stop_connector->conection_id = cid;
        struct hloop* ploop = get_hloop_from_index(index); //&hh_net.hoops[0];
        ASSERT(ploop != NULL);
        post_app_req(&ploop->loop, p_op_stop_connector);
    }

    return;
}

void stop_acceptor(const char* aid) {
    struct net_object_string_to_id* pfind = get_net_obejct(aid);
    if (pfind == NULL) {
        HLOG_ERROR("stop acceptor but failed to find it [%s]\n", aid);
        return;
    }
    struct op_stop_acceptor* p_op_stop_acceptor = create_op_stop_acceptor();
    ASSERT(p_op_stop_acceptor != NULL);
    p_op_stop_acceptor->conection_id = pfind->id;

    struct hloop* ploop = get_hloop(pfind->id);
    ASSERT(ploop != NULL);
    post_app_req(&ploop->loop, p_op_stop_acceptor);
}

void stop_connector(const char* cid) {
    struct net_object_string_to_id* pfind = get_net_obejct(cid);
    if (pfind == NULL) {
        HLOG_ERROR("stop connector but failed to find it [%s]\n", cid);
        return;
    }
    struct op_stop_connector* p_op_stop_connector = create_op_stop_connector();
    ASSERT(p_op_stop_connector != NULL);
    p_op_stop_connector->conection_id = pfind->id;
    struct hloop* ploop = get_hloop(pfind->id); //&hh_net.hoops[0];
    ASSERT(ploop != NULL);
    post_app_req(&ploop->loop, p_op_stop_connector);
}

static void get_net_event(QUEUE** qOut) {
    uv_mutex_lock(&hh_net.mutex_net_event);
    QUEUE_MOVE(&hh_net.net_event_list, *qOut);
    QUEUE_INIT(&hh_net.net_event_list);
    uv_mutex_unlock(&hh_net.mutex_net_event);
}

int run_hnet() {
    if (s_hnet_status != 1) {
        return -1;
    }
    QUEUE qevents;
    QUEUE_INIT(&qevents);
    QUEUE* head = &qevents;
    get_net_event(&head);
    int n = 0;
    QUEUE* p_tmp_item = NULL;
    while (!QUEUE_EMPTY(&qevents)) {
        p_tmp_item = QUEUE_NEXT(&qevents);
        QUEUE_REMOVE(p_tmp_item);
        net_app_base* p_app_base = QUEUE_DATA(p_tmp_item, net_app_base, event_node);
        if (p_app_base->net_app_type == E_NET_APP_ON_CONNECT_SUCCESS) {
            net_app_on_connect_success* p_on_connect_success = (net_app_on_connect_success*)(p_app_base);
            p_on_connect_success->ev_rv->ev(HNET_EV_ACT, p_on_connect_success->name, p_on_connect_success->net_object_id,
                                            p_on_connect_success->remote_ip, p_on_connect_success->remote_port,
                                            p_on_connect_success->local_ip, p_on_connect_success->local_port);
        } else if (p_app_base->net_app_type == E_NET_APP_ON_CONNECTION) {
            net_app_on_connection* p_on_connection = (net_app_on_connection*)(p_app_base);
            p_on_connection->ev_rv->ev(HNET_EV_PAS, p_on_connection->name, p_on_connection->net_object_id,
                                       p_on_connection->remote_ip, p_on_connection->remote_port,
                                       p_on_connection->local_ip, p_on_connection->local_port);
        } else if (p_app_base->net_app_type == E_NET_APP_ON_RECEIVE) {
            net_app_on_receive* p_on_receive = (net_app_on_receive*)(p_app_base);
            p_on_receive->ev_rv->rv(HNET_RV_GOOD, p_on_receive->net_object_id, p_on_receive->buf, p_on_receive->len);
        } else if (p_app_base->net_app_type == E_NET_APP_ON_CLOSE) {
            net_app_on_close* p_on_close = (net_app_on_close*)(p_app_base);
            p_on_close->ev_rv->rv(HNET_RV_BAD, p_on_close->net_object_id, NULL, 0);
        }
        n++;
        free(p_app_base);
    }
    if (n > 0) {
        HLOG_INFO("--------es[%d]--------\n", n);
    }
    return n;
}

void post_hb() {
    for (int i = 0; i < LOOP_SIZE(hh_net.hoops); i++) {
        struct op_hb* pop_hb = create_op_hb();
        UV_REQ_INIT(&pop_hb->op, UV_APP);
        pop_hb->op.loop = &hh_net.hoops[i].loop;
        post_app_req(&hh_net.hoops[i].loop, pop_hb);
    }
}
#define M_COUNT 100000000 
unsigned long long ggg = 0;
void test_malloc() {
    for (int i = 0; i < M_COUNT; i++) {
        int* p = (int*)malloc(sizeof(int));
        *p = i;
        ggg += *p;
        free(p);
    }
    printf("total %llu ", ggg);
}
